Experiment with redirecting all BrowserThreads (but UI/IO) to TaskScheduler

content/browser/browser_thread_impl.cc

Index: content/browser/browser_thread_impl.cc
diff --git a/content/browser/browser_thread_impl.cc b/content/browser/browser_thread_impl.cc
index 72345852c0f5e7e64f5c3d501839dda16748c5e5..641117d1cc2c65baa71cf47421811b67f5ed301a 100644
--- a/content/browser/browser_thread_impl.cc
+++ b/content/browser/browser_thread_impl.cc
@@ -4,18 +4,18 @@
 
 #include "content/browser/browser_thread_impl.h"
 
-#include <string.h>
-
 #include <string>
 
 #include "base/atomicops.h"
 #include "base/bind.h"
 #include "base/compiler_specific.h"
 #include "base/lazy_instance.h"
+#include "base/logging.h"
 #include "base/macros.h"
+#include "base/message_loop/message_loop.h"
 #include "base/profiler/scoped_tracker.h"
 #include "base/run_loop.h"
-#include "base/single_thread_task_runner.h"
+#include "base/synchronization/waitable_event.h"
 #include "base/threading/platform_thread.h"
 #include "base/threading/sequenced_worker_pool.h"
 #include "build/build_config.h"
@@ -52,6 +52,8 @@ static const char* GetThreadName(BrowserThread::ID thread) {
 
 // An implementation of SingleThreadTaskRunner to be used in conjunction
 // with BrowserThread.
+// TODO(gab): Consider replacing this with |g_globals->task_runners| -- only
+// works if none are requested before starting the threads.
 class BrowserThreadTaskRunner : public base::SingleThreadTaskRunner {
  public:
   explicit BrowserThreadTaskRunner(BrowserThread::ID identifier)
@@ -99,6 +101,20 @@ struct BrowserThreadTaskRunners {
 base::LazyInstance<BrowserThreadTaskRunners>::Leaky g_task_runners =
     LAZY_INSTANCE_INITIALIZER;
 
+// State of a given BrowserThread::ID in chronological order throughout the
+// browser process' lifetime.
+enum BrowserThreadState {
+  // BrowserThread::ID isn't associated with anything yet.
+  UNINITIALIZED = 0,
+  // BrowserThread::ID is associated with a BrowserThreadImpl instance but the
+  // underlying thread hasn't started yet.
+  INITIALIZED,
+  // BrowserThread::ID is associated to a TaskRunner and is accepting tasks.
+  RUNNING,
+  // BrowserThread::ID no longer accepts tasks.
+  SHUTDOWN
+};
+
 using BrowserThreadDelegateAtomicPtr = base::subtle::AtomicWord;
 
 struct BrowserThreadGlobals {
@@ -106,24 +122,21 @@ struct BrowserThreadGlobals {
       : blocking_pool(
             new base::SequencedWorkerPool(3,
                                           "BrowserBlocking",
-                                          base::TaskPriority::USER_VISIBLE)) {
-    memset(threads, 0, BrowserThread::ID_COUNT * sizeof(threads[0]));
-    memset(thread_ids, 0, BrowserThread::ID_COUNT * sizeof(thread_ids[0]));
-  }
+                                          base::TaskPriority::USER_VISIBLE)) {}
 
-  // This lock protects |threads| and |thread_ids|. Do not read or modify those
+  // This lock protects |task_runners| and |states|. Do not read or modify those
   // arrays without holding this lock. Do not block while holding this lock.
   base::Lock lock;
 
-  // This array is protected by |lock|. IDs in this array are populated as soon
-  // as their respective thread is started and are never reset.
-  base::PlatformThreadId thread_ids[BrowserThread::ID_COUNT];
+  // This array is filled either as the underlying threads start and invoke
+  // Init() or in RedirectThreadIDToTaskRunner() for threads that are being
+  // redirected. It is not emptied during shutdown in order to support
+  // RunsTasksOnCurrentThread() until the very end.
+  scoped_refptr<base::SingleThreadTaskRunner>
+      task_runners[BrowserThread::ID_COUNT];
 
-  // This array is protected by |lock|. The threads are not owned by this
-  // array. Typically, the threads are owned on the UI thread by
-  // BrowserMainLoop. BrowserThreadImpl objects remove themselves from this
-  // array upon destruction.
-  BrowserThreadImpl* threads[BrowserThread::ID_COUNT];
+  // Holds the state of each BrowserThread::ID.
+  BrowserThreadState states[BrowserThread::ID_COUNT] = {};
 
   // Only atomic operations are used on this pointer. The delegate isn't owned
   // by BrowserThreadGlobals, rather by whoever calls
@@ -141,13 +154,6 @@ base::LazyInstance<BrowserThreadGlobals>::Leaky
 BrowserThreadImpl::BrowserThreadImpl(ID identifier)
     : Thread(GetThreadName(identifier)), identifier_(identifier) {
   Initialize();
-
-  // Unit tests may create multiple TestBrowserThreadBundles, causing the same
-  // BrowserThread ID to be reinitialized. We explicitly clear the thread ID
-  // here so Start() can sanity check.
-  BrowserThreadGlobals& globals = g_globals.Get();
-  base::AutoLock lock(globals.lock);
-  globals.thread_ids[identifier] = base::kInvalidThreadId;
 }
 
 BrowserThreadImpl::BrowserThreadImpl(ID identifier,
@@ -156,11 +162,16 @@ BrowserThreadImpl::BrowserThreadImpl(ID identifier,
   SetMessageLoop(message_loop);
   Initialize();
 
-  // If constructed with an explicit message loop, this is a fake BrowserThread
-  // which runs on the current thread.
+  // If constructed with an explicit message loop, this is a fake
+  // BrowserThread which runs on the current thread.
   BrowserThreadGlobals& globals = g_globals.Get();
   base::AutoLock lock(globals.lock);
-  globals.thread_ids[identifier] = base::PlatformThread::CurrentId();
+
+  DCHECK(!globals.task_runners[identifier_]);
+  globals.task_runners[identifier_] = task_runner();
+
+  DCHECK_EQ(globals.states[identifier_], BrowserThreadState::INITIALIZED);
+  globals.states[identifier_] = BrowserThreadState::RUNNING;
 }
 
 // static
@@ -186,14 +197,27 @@ void BrowserThreadImpl::FlushThreadPoolHelperForTesting() {
 void BrowserThreadImpl::Init() {
   BrowserThreadGlobals& globals = g_globals.Get();
 
+#if DCHECK_IS_ON()
+  {
+    base::AutoLock lock(globals.lock);
+    // |globals| should already have been initialized for |identifier_| in
+    // BrowserThreadImpl::StartWithOptions(). If this isn't the case it's likely
+    // because this BrowserThreadImpl's owner incorrectly used Thread::Start.*()
+    // instead of BrowserThreadImpl::Start.*().
+    DCHECK_EQ(globals.states[identifier_], BrowserThreadState::RUNNING);
+    DCHECK(globals.task_runners[identifier_]);
+    DCHECK(globals.task_runners[identifier_]->RunsTasksOnCurrentThread());
+  }
+#endif  // DCHECK_IS_ON()
+
   if (identifier_ == BrowserThread::DB ||
       identifier_ == BrowserThread::FILE ||
       identifier_ == BrowserThread::FILE_USER_BLOCKING ||
       identifier_ == BrowserThread::PROCESS_LAUNCHER ||
       identifier_ == BrowserThread::CACHE) {
-    base::MessageLoop* message_loop = base::MessageLoop::current();
-    message_loop->DisallowNesting();
-    message_loop->DisallowTaskObservers();
+    // Nesting and task observers are not allowed on redirected threads.
+    message_loop()->DisallowNesting();
+    message_loop()->DisallowTaskObservers();
   }
 
   if (identifier_ == BrowserThread::IO) {
@@ -305,12 +329,13 @@ void BrowserThreadImpl::CleanUp() {
       reinterpret_cast<BrowserThreadDelegate*>(delegate)->CleanUp();
   }
 
-  // PostTaskHelper() accesses the message loop while holding this lock.
-  // However, the message loop will soon be destructed without any locking. So
-  // to prevent a race with accessing the message loop in PostTaskHelper(),
-  // remove this thread from the global array now.
+  // Change the state to SHUTDOWN so that PostTaskHelper stops accepting tasks
+  // for this thread. Do not clear globals.task_runners[identifier_] so that
+  // BrowserThread::CurrentlyOn() works from the MessageLoop's
+  // DestructionObservers.
   base::AutoLock lock(globals.lock);
-  globals.threads[identifier_] = nullptr;
+  DCHECK_EQ(globals.states[identifier_], BrowserThreadState::RUNNING);
+  globals.states[identifier_] = BrowserThreadState::SHUTDOWN;
 }
 
 void BrowserThreadImpl::Initialize() {
@@ -319,8 +344,19 @@ void BrowserThreadImpl::Initialize() {
   base::AutoLock lock(globals.lock);
   DCHECK_GE(identifier_, 0);
   DCHECK_LT(identifier_, ID_COUNT);
-  DCHECK_EQ(globals.threads[identifier_], nullptr);
-  globals.threads[identifier_] = this;
+  DCHECK_EQ(globals.states[identifier_], BrowserThreadState::UNINITIALIZED);
+  globals.states[identifier_] = BrowserThreadState::INITIALIZED;
+}
+
+// static
+void BrowserThreadImpl::ResetGlobalsForTesting(BrowserThread::ID identifier) {
+  BrowserThreadGlobals& globals = g_globals.Get();
+
+  base::AutoLock lock(globals.lock);
+  DCHECK_EQ(globals.states[identifier], BrowserThreadState::SHUTDOWN);
+  globals.states[identifier] = BrowserThreadState::UNINITIALIZED;
+  globals.task_runners[identifier] = nullptr;
+  SetDelegate(identifier, nullptr);
 }
 
 BrowserThreadImpl::~BrowserThreadImpl() {
@@ -331,12 +367,22 @@ BrowserThreadImpl::~BrowserThreadImpl() {
 
   BrowserThreadGlobals& globals = g_globals.Get();
   base::AutoLock lock(globals.lock);
-  globals.threads[identifier_] = nullptr;
-#ifndef NDEBUG
+  // This thread should have gone through Cleanup() as part of Stop() and be in
+  // the SHUTDOWN state already (unless it uses an externally provided
+  // MessageLoop instead of a real underlying thread and thus doesn't go through
+  // Cleanup()).
+  if (using_external_message_loop()) {
+    DCHECK_EQ(globals.states[identifier_], BrowserThreadState::RUNNING);
+    globals.states[identifier_] = BrowserThreadState::SHUTDOWN;
+  } else {
+    DCHECK_EQ(globals.states[identifier_], BrowserThreadState::SHUTDOWN);
+  }
+#if DCHECK_IS_ON()
   // Double check that the threads are ordered correctly in the enumeration.
   for (int i = identifier_ + 1; i < ID_COUNT; ++i) {
-    DCHECK(!globals.threads[i]) <<
-        "Threads must be listed in the reverse order that they die";
+    DCHECK(globals.states[i] == BrowserThreadState::SHUTDOWN ||
+           globals.states[i] == BrowserThreadState::UNINITIALIZED)
+        << "Threads must be listed in the reverse order that they die";
   }
 #endif
 }
@@ -346,14 +392,24 @@ bool BrowserThreadImpl::Start() {
 }
 
 bool BrowserThreadImpl::StartWithOptions(const Options& options) {
-  // The global thread table needs to be locked while a new thread is
-  // starting, as the new thread can asynchronously start touching the
-  // table (and other thread's message_loop).
   BrowserThreadGlobals& globals = g_globals.Get();
+
+  // Holding the lock is necessary when kicking off the thread to ensure
+  // |states| and |task_runners| are updated before it gets to query them.
   base::AutoLock lock(globals.lock);
-  DCHECK_EQ(globals.thread_ids[identifier_], base::kInvalidThreadId);
+
   bool result = Thread::StartWithOptions(options);
-  globals.thread_ids[identifier_] = GetThreadId();
+
+  // Although the thread is starting asynchronously. The MessageLoop is already

[fdoray, 2016/12/08 14:51:46]

Although the thread is starting asynchronously <<,>> the

[gab, 2016/12/08 18:49:41]

Done.

+  // ready to accept tasks and as such this BrowserThreadImpl is considered as
+  // "running".
+  DCHECK(!globals.task_runners[identifier_]);
+  globals.task_runners[identifier_] = task_runner();
+  DCHECK(globals.task_runners[identifier_]);
+
+  DCHECK_EQ(globals.states[identifier_], BrowserThreadState::INITIALIZED);
+  globals.states[identifier_] = BrowserThreadState::RUNNING;

[robliao, 2016/12/08 01:57:17]

In the previous code, if thread creation failed, we hang forever (GetThreadId
appears to wait for ThreadMain to signal its event).

In this case, we mark the thread as running and proceed. Is this okay?

[fdoray, 2016/12/08 14:51:46]

This change could make startup faster!

[gab, 2016/12/08 18:49:41]

Yes this is fine because browser_main_loop.cc does this:

      if (!(*thread_to_start)->StartWithOptions(options))
        LOG(FATAL) << "Failed to start the browser thread: id == " << id;
Exactly, I was surprised to see that we were blocking on getting the ID here,
the whole point of making the start asynchronous a year or two ago was to avoid
blocking on threads that are launching (i.e. it's fine to let them start async
so long as we make their ML/TaskRunners available the callers can't notice the
difference). Looks like this was an unintentional regression from
http://crrev.com/408295.

So we're back to an asynchronous thread start, yay :).

Updated CL desc to highlight this.

[robliao, 2016/12/14 20:38:13]

The state seems a little wonky given that we enter RUNNING even on failure, but
given we check on the outside, I think I can live with that.

+
   return result;
 }
 
@@ -363,6 +419,70 @@ bool BrowserThreadImpl::StartAndWaitForTesting() {
   WaitUntilThreadStarted();
   return true;
 }
+
+// static
+void BrowserThreadImpl::RedirectThreadIDToTaskRunner(
+    BrowserThread::ID identifier,
+    scoped_refptr<base::SingleThreadTaskRunner> task_runner) {
+  DCHECK(task_runner);
+
+  BrowserThreadGlobals& globals = g_globals.Get();
+  base::AutoLock lock(globals.lock);
+
+  DCHECK(!globals.task_runners[identifier]);
+  DCHECK_EQ(globals.states[identifier], BrowserThreadState::UNINITIALIZED);
+
+  globals.task_runners[identifier] = std::move(task_runner);
+  globals.states[identifier] = BrowserThreadState::RUNNING;
+}
+
+// static
+void BrowserThreadImpl::StopRedirectionOfThreadID(
+    BrowserThread::ID identifier) {
+  BrowserThreadGlobals& globals = g_globals.Get();
+  base::AutoLock auto_lock(globals.lock);
+
+  DCHECK(globals.task_runners[identifier]);
+
+  // Change the state to SHUTDOWN to stop accepting new tasks. Note: this is
+  // different from non-redirected threads which continue accepting tasks while
+  // being joined and only quit when idle. However, any tasks for which this
+  // difference matters was already racy as any thread posting a task after the
+  // Signal task below can't be synchronized with the joining thread. Therefore,
+  // that task could already come in before or after the join had completed in
+  // the non-redirection world. Entering SHUTDOWN early merely skews this race
+  // towards making it less likely such a task is accepted by the joined thread
+  // which is fine.
+  DCHECK_EQ(globals.states[identifier], BrowserThreadState::RUNNING);
+  globals.states[identifier] = BrowserThreadState::SHUTDOWN;
+
+  // Wait for all pending tasks to complete.
+  base::WaitableEvent flushed(base::WaitableEvent::ResetPolicy::MANUAL,
+                              base::WaitableEvent::InitialState::NOT_SIGNALED);
+  globals.task_runners[identifier]->PostTask(
+      FROM_HERE,
+      base::Bind(&base::WaitableEvent::Signal, base::Unretained(&flushed)));
+  {
+    base::AutoUnlock auto_lock(globals.lock);
+    flushed.Wait();
+  }
+
+  // Only reset the task runner after running pending tasks so that
+  // BrowserThread::CurrentlyOn() works in their scope.
+  globals.task_runners[identifier] = nullptr;
+
+  // Note: it's still possible for tasks to be posted to that task runner after
+  // this point (e.g. through a previously obtained ThreadTaskRunnerHandle or by
+  // one of the last tasks re-posting to its ThreadTaskRunnerHandle) but the
+  // BrowserThread API itself won't accept tasks. Such tasks are ultimately
+  // guaranteed to run before TaskScheduler::Shutdown() returns but may break
+  // the assumption in PostTaskHelper that BrowserThread::ID A > B will always
+  // succeed to post to B. This is pretty much the only observable difference
+  // between a redirected thread and a real one and is one we're willing to live
+  // with for this experiment. TODO(gab): fix this before enabling the
+  // experiment by default on trunk, http://crbug.com/653916.
+}
+
 // static
 bool BrowserThreadImpl::PostTaskHelper(
     BrowserThread::ID identifier,
@@ -374,7 +494,8 @@ bool BrowserThreadImpl::PostTaskHelper(
   DCHECK_LT(identifier, ID_COUNT);
   // Optimization: to avoid unnecessary locks, we listed the ID enumeration in
   // order of lifetime.  So no need to lock if we know that the target thread
-  // outlives current thread.
+  // outlives current thread as that implies the current thread only ever sees
+  // the target thread in its RUNNING state.
   // Note: since the array is so small, ok to loop instead of creating a map,
   // which would require a lock because std::map isn't thread safe, defeating
   // the whole purpose of this optimization.
@@ -387,22 +508,23 @@ bool BrowserThreadImpl::PostTaskHelper(
   if (!target_thread_outlives_current)
     globals.lock.Acquire();
 
-  base::MessageLoop* message_loop =
-      globals.threads[identifier] ? globals.threads[identifier]->message_loop()
-                                  : nullptr;
-  if (message_loop) {
+  const bool accepting_tasks =
+      globals.states[identifier] == BrowserThreadState::RUNNING;
+  if (accepting_tasks) {
+    base::SingleThreadTaskRunner* task_runner =

[robliao, 2016/12/08 01:57:17]

Since we don't always lock here, it seems like it's possible for a task to get
posted after the flush event signal task above. Maybe we should lock now?

[gab, 2016/12/08 18:49:41]

There is indeed a theoretical race as described in
BrowserThreadImpl::StopRedirectionOfThreadID() because a task posted to a
flushed thread A can theoretically run after a thread B that is meant to die
after thread A and B can thus unexpectedly have been shutdown ahead of a task on
A.

This theoretical race is only there on shutdown when the experiment is enabled.
The better solution long-term is to join IO thread, then
TaskScheduler::Shutdown() then destroy BrowserThreadImpl for UI thread. But this
requires that all other threads be redirected or we still have ordering
problems.

Solving the race in the meantime either requires always locking here (which
breaks my intent of this CL being status quo) or truly mimic'ing status quo by
introducing a Join on task_scheduler::Sequence which is a pain for something we
don't want long-term.

So we figured we were okay with a theoretical shutdown race in the initial
experiment (there's a TODO in BrowserThreadImpl::StopRedirectionOfThreadID() to
fix this before making it default).

+        globals.task_runners[identifier].get();
+    DCHECK(task_runner);
     if (nestable) {
-      message_loop->task_runner()->PostDelayedTask(from_here, task, delay);
+      task_runner->PostDelayedTask(from_here, task, delay);
     } else {
-      message_loop->task_runner()->PostNonNestableDelayedTask(from_here, task,
-                                                              delay);
+      task_runner->PostNonNestableDelayedTask(from_here, task, delay);
     }
   }
 
   if (!target_thread_outlives_current)
     globals.lock.Release();
 
-  return !!message_loop;
+  return accepting_tasks;
 }
 
 // static
@@ -453,7 +575,8 @@ bool BrowserThread::IsThreadInitialized(ID identifier) {
   base::AutoLock lock(globals.lock);
   DCHECK_GE(identifier, 0);
   DCHECK_LT(identifier, ID_COUNT);
-  return globals.threads[identifier] != nullptr;
+  return globals.states[identifier] == BrowserThreadState::INITIALIZED ||
+         globals.states[identifier] == BrowserThreadState::RUNNING;
 }
 
 // static
@@ -462,7 +585,8 @@ bool BrowserThread::CurrentlyOn(ID identifier) {
   base::AutoLock lock(globals.lock);
   DCHECK_GE(identifier, 0);
   DCHECK_LT(identifier, ID_COUNT);
-  return base::PlatformThread::CurrentId() == globals.thread_ids[identifier];
+  return globals.task_runners[identifier] &&
+         globals.task_runners[identifier]->RunsTasksOnCurrentThread();
 }
 
 // static
@@ -488,8 +612,7 @@ bool BrowserThread::IsMessageLoopValid(ID identifier) {
   base::AutoLock lock(globals.lock);
   DCHECK_GE(identifier, 0);
   DCHECK_LT(identifier, ID_COUNT);
-  return globals.threads[identifier] &&
-         globals.threads[identifier]->message_loop();
+  return globals.states[identifier] == BrowserThreadState::RUNNING;
 }
 
 // static
@@ -543,7 +666,6 @@ bool BrowserThread::GetCurrentThreadIdentifier(ID* identifier) {
   if (g_globals == nullptr)
     return false;
 
-  base::MessageLoop* cur_message_loop = base::MessageLoop::current();
   BrowserThreadGlobals& globals = g_globals.Get();
   // Profiler to track potential contention on |globals.lock|. This only does
   // real work on canary and local dev builds, so the cost of having this here
@@ -551,9 +673,9 @@ bool BrowserThread::GetCurrentThreadIdentifier(ID* identifier) {
   tracked_objects::ScopedTracker tracking_profile(FROM_HERE);
   base::AutoLock lock(globals.lock);
   for (int i = 0; i < ID_COUNT; ++i) {
-    if (globals.threads[i] &&
-        globals.threads[i]->message_loop() == cur_message_loop) {
-      *identifier = globals.threads[i]->identifier_;
+    if (globals.task_runners[i] &&
+        globals.task_runners[i]->RunsTasksOnCurrentThread()) {
+      *identifier = static_cast<ID>(i);
       return true;
     }
   }