TaskScheduler [7] SchedulerThreadPool

base/task_scheduler/scheduler_thread_pool.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_SCHEDULER_THREAD_POOL_H_
+#define BASE_TASK_SCHEDULER_SCHEDULER_THREAD_POOL_H_
+
+#include <stddef.h>
+
+#include <memory>
+#include <stack>
+#include <vector>
+
+#include "base/base_export.h"
+#include "base/callback.h"
+#include "base/macros.h"
+#include "base/memory/ref_counted.h"
+#include "base/synchronization/condition_variable.h"
+#include "base/task_runner.h"
+#include "base/task_scheduler/priority_queue.h"
+#include "base/task_scheduler/scheduler_lock.h"
+#include "base/task_scheduler/scheduler_worker_thread.h"
+#include "base/task_scheduler/scheduler_worker_thread_delegate.h"
+#include "base/task_scheduler/sequence.h"
+#include "base/task_scheduler/task_traits.h"
+#include "base/threading/platform_thread.h"
+
+namespace base {
+namespace internal {
+
+struct SequenceSortKey;
+class TaskTracker;
+
+// A pool of threads that run Tasks. This class is thread-safe.
+class BASE_EXPORT SchedulerThreadPool : public SchedulerWorkerThreadDelegate {
+ public:
+  // Callback invoked after a worker thread has tried to run a Task from
+  // |sequence| (a TaskTracker might have prevented the Task from running).
+  using RanTaskFromSequenceCallback = Callback<void(scoped_refptr<Sequence>)>;
+
+  // Creates a SchedulerThreadPool with up to |max_threads| threads of priority
+  // |thread_priority|. |ran_task_from_sequence_callback| will be invoked after
+  // a thread of this thread pool tries to run a Task. |task_tracker| is used to
+  // handle shutdown behavior of Tasks. Returns nullptr when it's not possible
+  // to create a thread pool with at least one thread.

[gab, 2016/04/07 20:32:48]

s/when it's not possible to/on failure to/

[fdoray, 2016/04/08 14:53:03]

Done.

+  static std::unique_ptr<SchedulerThreadPool> CreateThreadPool(
+      ThreadPriority thread_priority,
+      size_t max_threads,
+      const RanTaskFromSequenceCallback& ran_task_from_sequence_callback,
+      TaskTracker* task_tracker);
+
+  // Destroying a SchedulerThreadPool returned by CreateThreadPool() is not
+  // allowed in production; it is always leaked. In tests, it can only be
+  // destroyed after JoinForTesting() has returned.
+  ~SchedulerThreadPool();
+
+  // Returns a TaskRunner whose PostTask invocations will result in scheduling
+  // Tasks with |traits| and |execution_mode| in this thread pool.

[gab, 2016/04/07 20:32:48]

Does anything make sure that this STP is the right pool for these |traits| or is
that up to the caller to decide?

[fdoray, 2016/04/08 14:53:03]

It's up to the caller to decide. The mapping TaskTraits -> ThreadPool will live
in TaskSchedulerImpl. If we add checks in ThreadPool, we'll have to update
multiple components when we decide to change the mapping.

+  scoped_refptr<TaskRunner> CreateTaskRunnerWithTraits(
+      const TaskTraits& traits,
+      ExecutionMode execution_mode);
+
+  // Inserts |sequence| into this thread pool's shared priority queue with
+  // |sequence_sort_key|. Must only be used by a worker thread to put |sequence|

[gab, 2016/04/07 20:32:48]

I would assume SWT doesn't have a pointer back to its STP and that this is
instead bound into a callback and that this can thus be priavte (actually isn't
this RanTaskFromSequence() -- I'm confused on the difference here, is it because
half the decision is made outside and loops back into
InsertSequenceAfterTaskRan() if it should be re-inserted in this ThreadPool?)

[fdoray, 2016/04/08 14:53:03]

Currently, the stack to reinsert a Sequence in a shared PriorityQueue after a
pop is:

1. SchedulerWorkerThread::ThreadMain
2. SchedulerThreadPool::EnqueueSequence (delegate)
3. TaskSchedulerImpl::EnqueueSequenceAfterTaskRan (enqueue_sequence_callback_)
[1]
4. SchedulerThreadPool::EnqueueSequenceAfterTaskRan [2]

[1] Computes the Sequence's sort key and determines in which PQ it should be
reinserted.
[2] The SchedulerThreadPool isn't necessarily the same as in 2.

When WorkerThread had multiple callbacks instead of a delegate, 1 invoked 3
directly and we didn't have 2 methods with similar names on ThreadPool (the
names are very bad currently). Can I remove EnqueueSequence from
SchedulerWorkerThread::Delegate and use a callback to go directly from 1 to 3?

+  // back into a PriorityQueue after running a Task from it.
+  void InsertSequenceAfterTaskRan(scoped_refptr<Sequence> sequence,
+                                  const SequenceSortKey& sequence_sort_key);
+
+  // Waits until all threads are idle.
+  void WaitForAllWorkerThreadsIdleForTesting();
+
+  // Joins all threads of this thread pool. Tasks that are already running are
+  // allowed to complete their execution. This can only be called once.
+  void JoinForTesting();
+
+ private:
+  SchedulerThreadPool(
+      const RanTaskFromSequenceCallback& ran_task_from_sequence_callback,
+      TaskTracker* task_tracker);
+
+  void Initialize(ThreadPriority thread_priority, size_t max_threads);
+
+  // Wakes up one thread from this thread pool if they aren't all busy.

[gab, 2016/04/07 20:32:47]

How about:

// Wakes up the last thread from this thread pool to go idle, if any.

[fdoray, 2016/04/08 14:53:02]

Done.

+  void WakeUpOneThread();
+
+  // Adds |worker_thread| to |idle_worker_threads_stack_|.
+  void AddToIdleWorkerThreadsStack(SchedulerWorkerThread* worker_thread);
+
+  // SchedulerWorkerThreadDelegate:
+  void OnMainEntry() override;
+  void OnMainExit() override;
+  scoped_refptr<Sequence> GetWork(
+      SchedulerWorkerThread* worker_thread) override;
+  void RanTaskFromSequence(scoped_refptr<Sequence> sequence) override;
+
+  // PriorityQueue from which all threads of this thread pool get work.
+  PriorityQueue shared_priority_queue_;
+
+  // All worker threads owned by this thread pool. This is only modified by the
+  // constructor.

[gab, 2016/04/07 20:32:48]

I don't think this last requirement "This is only modified by the constructor"
is necessary. I definitely see a future world where the list of worker_threads_
will be dynamic. Doesn't need to be now but no need to add this API explicit
requirement and restrict future evolutions of the impl.

[fdoray, 2016/04/08 14:53:02]

Done. Without this comment, I have to hold a lock to access |worker_threads_|.

[gab, 2016/04/08 17:56:00]

Good point, but I think that's the right thing to do (we'll have to long term
anyways).

[fdoray, 2016/04/08 19:00:05]

Ack.

+  std::vector<std::unique_ptr<SchedulerWorkerThread>> worker_threads_;
+
+  // Synchronizes access to |idle_worker_threads_stack_| and
+  // |idle_worker_threads_cv_|.

[gab, 2016/04/07 20:32:48]

Add a comment explaining why it takes the shared PQ lock as a predecessor.

[fdoray, 2016/04/08 14:53:03]

Done.

+  SchedulerLock idle_worker_threads_stack_lock_;
+
+  // Stack of idle worker threads. The last worker thread that became idle is on
+  // top of the stack.

[gab, 2016/04/07 20:32:48]

rm "The last worker thread that became idle is top of the stack."

IMO this is implicit in the definition of a stack (where it's worth highlighting
IMO is on the APIs above as requested there).

[fdoray, 2016/04/08 14:53:03]

Done.

+  std::stack<SchedulerWorkerThread*> idle_worker_threads_stack_;
+
+  // Signaled when all worker threads become idle.
+  std::unique_ptr<ConditionVariable> idle_worker_threads_stack_cv_;

[gab, 2016/04/07 20:32:48]

Should we brand this with "for_testing"? I was confused why this was required
until I realized it was test-only.

We should at least have its comment mention this?

[danakj, 2016/04/07 23:08:43]

+1

[fdoray, 2016/04/08 14:53:03]

Done.

+
+  // Synchronizes access to |join_for_testing_returned_|.
+  SchedulerLock join_for_testing_returned_lock_;
+
+  // True once JoinForTesting() has returned.
+  bool join_for_testing_returned_ = false;
+
+  const RanTaskFromSequenceCallback ran_task_from_sequence_callback_;
+  TaskTracker* const task_tracker_;
+
+  DISALLOW_COPY_AND_ASSIGN(SchedulerThreadPool);
+};
+
+// Helper for posting |task| to the provided |sequence| and |priority_queue|
+// conditional on |task_tracker|. Returns true if |task| is posted. Exposed for
+// testing.
+bool BASE_EXPORT PostTaskHelper(std::unique_ptr<Task> task,
+                                scoped_refptr<Sequence> sequence,
+                                PriorityQueue* priority_queue,
+                                TaskTracker* task_tracker);
+
+}  // namespace internal
+}  // namespace base
+
+#endif  // BASE_TASK_SCHEDULER_SCHEDULER_THREAD_POOL_H_