Virtualize The Existence of a Scheduler Worker Thread

base/task_scheduler/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_SCHEDULER_WORKER_THREAD_H_
 #define BASE_TASK_SCHEDULER_SCHEDULER_WORKER_THREAD_H_
 
 #include <memory>
 
 #include "base/base_export.h"
 #include "base/macros.h"
 #include "base/memory/ref_counted.h"
 #include "base/synchronization/waitable_event.h"
 #include "base/task_scheduler/scheduler_lock.h"
 #include "base/task_scheduler/sequence.h"
 #include "base/threading/platform_thread.h"
 #include "base/time/time.h"
 
 namespace base {
 namespace internal {
 
 class TaskTracker;
 
-// A thread that runs Tasks from Sequences returned by a delegate.
+// A worker that manages at most one thread to run Tasks from Sequences returned

[fdoray, 2016/06/13 20:31:17]

one thread at a time?

[robliao, 2016/06/13 23:23:28]

Changed to a single thread.

+// by a delegate.
 //
 // A SchedulerWorkerThread starts out sleeping. It is woken up by a call to
 // WakeUp(). After a wake-up, a SchedulerWorkerThread runs Tasks from Sequences
 // returned by the GetWork() method of its delegate as long as it doesn't return
 // nullptr. It also periodically checks with its TaskTracker whether shutdown
 // has completed and exits when it has.
 //
+// The worker is free to release and reallocate the platform thread with
+// guidance from the delegate.
+//
 // This class is thread-safe.
-class BASE_EXPORT SchedulerWorkerThread : public PlatformThread::Delegate {
+class BASE_EXPORT SchedulerWorkerThread {
  public:
   // Delegate interface for SchedulerWorkerThread. The methods are always called
   // from the thread managed by the SchedulerWorkerThread instance.
   class Delegate {
    public:
     virtual ~Delegate() = default;
 
-    // Called by |worker_thread| when it enters its main function.
+    // Called when |worker_thread|'s worker enters the main function.
+    // If a thread is recreated after detachment, this call will occur again.
     virtual void OnMainEntry(SchedulerWorkerThread* worker_thread) = 0;
 
     // Called by |worker_thread| to get a Sequence from which to run a Task.
     virtual scoped_refptr<Sequence> GetWork(
         SchedulerWorkerThread* worker_thread) = 0;
 
     // Called when |sequence| isn't empty after the SchedulerWorkerThread pops a
     // Task from it. |sequence| is the last Sequence returned by GetWork().
     virtual void ReEnqueueSequence(scoped_refptr<Sequence> sequence) = 0;
 
     // Called by |worker_thread| to determine how long to sleep before the next
     // call to GetWork(). GetWork() may be called before this timeout expires
     // if the thread's WakeUp() method is called.
     virtual TimeDelta GetSleepTimeout() = 0;
+
+    // Called by |worker_thread| to see if it is allowed to detach if the last
+    // call to GetWork() returned nullptr.

[fdoray, 2016/06/13 20:31:17]

No new line?

[robliao, 2016/06/13 23:23:28]

Done.

+    // A SchedulerWorkerThread reserves the right to ignore the return value of
+    // this function.

[fdoray, 2016/06/13 20:31:17]

How will the thread pool keep one active (not detached) idle thread at all time?

Could we change this comment to "From the time this returns true to the time
OnMainEntry() is called, the worker is in a detached state." In addition to the
idle stack, the thread pool would have a detached stack. CanDetach() would
return false if idle_stack_.size() <= 1. Before returning true, CanDetach()
would move the worker from idle_stack_ to detached_stack_. OnMainEntry() would
move the worker back to idle_stack_.

SchedulerWorkerThread::Detach() would always succeed. In addition to the
detached thread, it would return a boolean indicating whether a wake up was
pending at detachment time (IsWakeUpPending() read within the scope of
|worker_lock_|). If a wake up was pending at detachment time, call
outer_->WakeUp() just before setting |outer_| to nullptr.

[robliao, 2016/06/13 23:23:28]

The threadpool will simply make sure it returns false for CanDetach for at least
one worker. If there are no threads on the idle stack, it means all of them are
awake and running. It won't return true until there are some threads on the idle
stack.
Not sure we can. Calling OnMainEntry presumes that a worker is alive. It would
be strange to detach immediately after OnMainEntry.
The existing setup reduces thread churn. I'm a little suspicious if a thread is
going away and it itself requests a WakeUp. The natural question is if it's
trying to WakeUp a new thread, why not use the existing thread?

Having the thread request a WakeUp on a successful detachment would mean
shutting down the thread to create a new one.

+    //
+    // It is the responsibility of the delegate to determine if detachment is
+    // safe. If |worker_thread| is responsible for thread-affine work,
+    // detachment is generally not safe.
+    //
+    // When true is returned...

[gab, 2016/06/13 19:23:06]

s/.../:/

[robliao, 2016/06/13 23:23:28]

Done.

+    // - The next WakeUp() could be more costly due to new thread creation.
+    // - The worker will free resources if it can.

[gab, 2016/06/13 19:23:05]

The comment further above initially says "reserves the right to ignore" and here
it says "will free if it can"

[robliao, 2016/06/13 23:23:28]

Updated this comment and removed the statement above.

+    // False must be returned if SchedulerWorkerThread::JoinForTesting is in

[gab, 2016/06/13 19:23:06]

s/False must be returned/This *must* return false if SWT::JoinForTesting() is in
progress/

[robliao, 2016/06/13 23:23:28]

Done.

+    // progress.
+    virtual bool CanDetach(SchedulerWorkerThread* worker_thread) = 0;
   };
 
+  enum class InitialState { ALIVE, DETACHED };
+
   // Creates a SchedulerWorkerThread with priority |thread_priority| that runs
   // Tasks from Sequences returned by |delegate|. |task_tracker| is used to
-  // handle shutdown behavior of Tasks. Returns nullptr if creating the
-  // underlying platform thread fails.
+  // handle shutdown behavior of Tasks. If |worker_state| is DETACHED, the
+  // worker will be created upon a WakeUp. Returns nullptr if creating the
+  // underlying platform thread fails during Create.
   static std::unique_ptr<SchedulerWorkerThread> Create(
       ThreadPriority thread_priority,
       std::unique_ptr<Delegate> delegate,
-      TaskTracker* task_tracker);
+      TaskTracker* task_tracker,
+      InitialState initial_state);
 
   // Destroying a SchedulerWorkerThread in production is not allowed; it is
   // always leaked. In tests, it can only be destroyed after JoinForTesting()
   // has returned.
-  ~SchedulerWorkerThread() override;
+  ~SchedulerWorkerThread();
 
   // Wakes up this SchedulerWorkerThread if it wasn't already awake. After this
   // is called, this SchedulerWorkerThread will run Tasks from Sequences
   // returned by the GetWork() method of its delegate until it returns nullptr.
+  // May fail if the worker is detached. On failure the worker will not be able

[gab, 2016/06/13 19:23:06]

"the worker will not be able to call" : unclear for a reader IMO whether this
means the GetWork() call will fail or that the worker will be incapacitated and
not able to call it himself.

I still find the last two sentences confusing.

[robliao, 2016/06/13 23:23:28]

fdoray wanted to emphasize that WakeUp is generally a signal to call the
Delegate's GetWork. Upon failure to create a thread, GetWork will not be called.

Changed the wording to make this a little clearer.

+  // to call Delegate::GetWork().
   void WakeUp();
 
   SchedulerWorkerThread::Delegate* delegate() { return delegate_.get(); }
 
   // Joins this SchedulerWorkerThread. If a Task is already running, it will be
   // allowed to complete its execution. This can only be called once.
   void JoinForTesting();
 
+  // Returns ture if the worker is alive.

[fdoray, 2016/06/13 20:31:17]

*true

[robliao, 2016/06/13 23:23:28]

Done.

+  bool WorkerAliveForTesting() const;
+
  private:
+  class Worker;
+
   SchedulerWorkerThread(ThreadPriority thread_priority,
                         std::unique_ptr<Delegate> delegate,
                         TaskTracker* task_tracker);
 
-  // PlatformThread::Delegate:
-  void ThreadMain() override;
+  // Returns the worker instance if the detach was successful so that the worker
+  // instance can be freed upon termination of the thread.
+  // If the detach is not possible, returns nullptr.
+  std::unique_ptr<SchedulerWorkerThread::Worker> Detach();
+
+  void CreateWorker();
+
+  void CreateWorkerAssertSynchronized();
 
   bool ShouldExitForTesting() const;
 
-  // Platform thread managed by this SchedulerWorkerThread.
-  PlatformThreadHandle thread_handle_;
+  // Synchronizes access to |worker_|
+  mutable SchedulerLock worker_lock_;
 
-  // Event signaled to wake up this SchedulerWorkerThread.
-  WaitableEvent wake_up_event_;
+  // The underlying thread for this SchedulerWorkerThread.
+  std::unique_ptr<Worker> worker_;
 
+  const ThreadPriority thread_priority_;
   const std::unique_ptr<Delegate> delegate_;
   TaskTracker* const task_tracker_;
 
   // Synchronizes access to |should_exit_for_testing_|.
   mutable SchedulerLock should_exit_for_testing_lock_;
 
   // True once JoinForTesting() has been called.
   bool should_exit_for_testing_ = false;
 
   DISALLOW_COPY_AND_ASSIGN(SchedulerWorkerThread);
 };
 
 }  // namespace internal
 }  // namespace base
 
 #endif  // BASE_TASK_SCHEDULER_SCHEDULER_WORKER_THREAD_H_