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

content/browser/browser_main_loop.cc

 // Copyright (c) 2012 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.
 
 #include "content/browser/browser_main_loop.h"
 
 #include <stddef.h>
 
 #include <string>
 #include <utility>
@@ skipping 12 matching lines @@
 #include "base/metrics/histogram_macros.h"
 #include "base/metrics/user_metrics.h"
 #include "base/pending_task.h"
 #include "base/power_monitor/power_monitor.h"
 #include "base/power_monitor/power_monitor_device_source.h"
 #include "base/process/process_metrics.h"
 #include "base/run_loop.h"
 #include "base/single_thread_task_runner.h"
 #include "base/strings/string_number_conversions.h"
 #include "base/strings/string_split.h"
+#include "base/synchronization/waitable_event.h"
 #include "base/system_monitor/system_monitor.h"
+#include "base/task_scheduler/post_task.h"
+#include "base/task_scheduler/task_traits.h"
 #include "base/threading/sequenced_worker_pool.h"
 #include "base/threading/thread_restrictions.h"
 #include "base/threading/thread_task_runner_handle.h"
 #include "base/timer/hi_res_timer_manager.h"
 #include "base/trace_event/memory_dump_manager.h"
 #include "base/trace_event/trace_event.h"
 #include "build/build_config.h"
 #include "components/discardable_memory/service/discardable_shared_memory_manager.h"
 #include "components/tracing/browser/trace_config_file.h"
 #include "components/tracing/common/process_metrics_memory_dump_provider.h"
@@ skipping 250 matching lines @@
 void OnStoppedStartupTracing(const base::FilePath& trace_file) {
   VLOG(0) << "Completed startup tracing to " << trace_file.value();
 }
 
 // Disable optimizations for this block of functions so the compiler doesn't
 // merge them all together. This makes it possible to tell what thread was
 // unresponsive by inspecting the callstack.
 MSVC_DISABLE_OPTIMIZE()
 MSVC_PUSH_DISABLE_WARNING(4748)
 
-NOINLINE void ResetThread_DB(std::unique_ptr<BrowserProcessSubThread> thread) {
+// Takes ownership of |task_runner| and blocks until it's empty.
+void FlushThreadTaskRunner(
+    scoped_refptr<base::SingleThreadTaskRunner> task_runner) {
+  base::WaitableEvent thread_terminated(
+      base::WaitableEvent::ResetPolicy::MANUAL,
+      base::WaitableEvent::InitialState::NOT_SIGNALED);
+  task_runner->PostTask(FROM_HERE,
+                        base::Bind(&base::WaitableEvent::Signal,
+                                   base::Unretained(&thread_terminated)));
+  thread_terminated.Wait();
+}
+
+NOINLINE void ResetThread_DB(
+    std::unique_ptr<BrowserProcessSubThread> thread,
+    scoped_refptr<base::SingleThreadTaskRunner> task_runner) {
+  DCHECK((thread && !task_runner) || (!thread && task_runner));
   volatile int inhibit_comdat = __LINE__;
   ALLOW_UNUSED_LOCAL(inhibit_comdat);
-  thread.reset();
+  if (thread) {
+    thread.reset();
+  } else {

[fdoray, 2016/12/01 15:15:21]

You could avoid keeping SingleThreadTaskRunners references in BrowserMainLoop by
changing this code to:

auto task_runner = BrowserThreadImpl::RedirectThreadIDToTaskRunner(
    BrowserThread::DB);
FlushThreadTaskRunner(std::move(task_runner));

(assuming that RedirectThreadIDToTaskRunner() returns the previous TaskRunner
for the provided identifier)

[gab, 2016/12/07 19:15:27]

I was inspired by this and went a step further, the flush is now in
BrowserThreadImpl::StopRedirectionOfThreadID()

+    BrowserThreadImpl::RedirectThreadIDToTaskRunner(BrowserThread::DB, nullptr);

[fdoray, 2016/12/01 15:15:21]

Without redirection:
1. Main thread: ~BrowserProcessSubThread
2. Main thread: Thread::Stop()
3. Main thread: A QuitWhenIdle() task is posted to the thread.
4. Browser thread: Runs tasks until there are no more ready tasks.
5. Browser thread: Thread::Cleanup()
6. Browser thread: BrowserThreadImpl::CleanUp() sets the state to SHUTDOWN. No
more tasks can be posted to the thread.
7. Browser thread: Exits, unblocking the join call on the main thread.

With redirection:
1. Main thread: RedirectThreadIDToTaskRunner() sets the state to SHUTDOWN. No
more tasks can be posted to the thread via the BrowserThread API.
2. Main thread: FlushThreadTaskRunner() posts a WaitableEvent::Signal() task to
the browser thread and waits.
3. Browser thread: Runs all tasks that were posted before the call to
FlushThreadTaskRunner() and unblocks the main thread.

Without redirection, we wait for pending tasks to run and then we set the state
to SHUTDOWN. With redirection, we set the state to SHUTDOWN and then we wait for
tasks to run. Could we improve that?

[gab, 2016/12/07 19:15:26]

The thing is we have to change the state to SHUTDOWN in Cleanup() because it
always happens when the thread is stopped (which can happen outside of
~BrowserThreadImpl()) and since I want all state changes to DCHECK the previous
state for documentation it can't also be toggled to SHUTDOWN before invoking
Stop() in ~BrowserThreadImpl() -- and vice-versa, the thread could be SHUTDOWN
before ~BrowserThreadImpl().

Thinking about this some more I think it's fine as-is as:
1) Any tasks for which this matters is already racing in today's world as it has
to be posted from a thread which can't be aware of the main thread's state
(beyond, potentially, the fact that it's past MainMessageLoopRun()).
i.e. such a task already gets denied a %age of the time as it's not always
posted by the time the Join() starts and has therefore no guarantees of being
posted before the Join() completes. So triggering the SHUTDOWN state earlier
merely skews this race towards denying more of these tasks which seems fine to
me.

The WaitableEvent::Signal still matters though as it guarantees all tasks that
were knowingly scheduled ahead of the Join do get to run before the Join
completes.

As discussed further offline, there is still a minor edge case where previously
it was impossible  for a thread lower in BrowserThread::ID to fail to post to a
higher thread. Whereas there is now a race (e.g. it's possible for a
PostTaskAndReply to come back to the "shutdown" thread and run after the "join"
-- before the final TaskScheduler::Shutdown()). We have a solution for this but
won't implement it in this CL as the race only exists behind the experiment and
we'll fix it before making it default on trunk.

Added comments to this effect in BrowserThreadImpl::StopRedirectionToThreadID().

+    FlushThreadTaskRunner(std::move(task_runner));
+  }
 }
 
 NOINLINE void ResetThread_FILE(
-    std::unique_ptr<BrowserProcessSubThread> thread) {
+    std::unique_ptr<BrowserProcessSubThread> thread,
+    scoped_refptr<base::SingleThreadTaskRunner> task_runner) {
+  DCHECK((thread && !task_runner) || (!thread && task_runner));
   volatile int inhibit_comdat = __LINE__;
   ALLOW_UNUSED_LOCAL(inhibit_comdat);
-  thread.reset();
+  if (thread) {
+    thread.reset();
+  } else {
+    BrowserThreadImpl::RedirectThreadIDToTaskRunner(BrowserThread::FILE,
+                                                    nullptr);
+    FlushThreadTaskRunner(std::move(task_runner));
+  }
 }
 
 NOINLINE void ResetThread_FILE_USER_BLOCKING(
-    std::unique_ptr<BrowserProcessSubThread> thread) {
+    std::unique_ptr<BrowserProcessSubThread> thread,
+    scoped_refptr<base::SingleThreadTaskRunner> task_runner) {
+  DCHECK((thread && !task_runner) || (!thread && task_runner));
   volatile int inhibit_comdat = __LINE__;
   ALLOW_UNUSED_LOCAL(inhibit_comdat);
-  thread.reset();
+  if (thread) {
+    thread.reset();
+  } else {
+    BrowserThreadImpl::RedirectThreadIDToTaskRunner(
+        BrowserThread::FILE_USER_BLOCKING, nullptr);
+    FlushThreadTaskRunner(std::move(task_runner));
+  }
 }
 
 NOINLINE void ResetThread_PROCESS_LAUNCHER(
-    std::unique_ptr<BrowserProcessSubThread> thread) {
+    std::unique_ptr<BrowserProcessSubThread> thread,
+    scoped_refptr<base::SingleThreadTaskRunner> task_runner) {
+  DCHECK((thread && !task_runner) || (!thread && task_runner));
   volatile int inhibit_comdat = __LINE__;
   ALLOW_UNUSED_LOCAL(inhibit_comdat);
-  thread.reset();
+  if (thread) {
+    thread.reset();
+  } else {
+    BrowserThreadImpl::RedirectThreadIDToTaskRunner(
+        BrowserThread::PROCESS_LAUNCHER, nullptr);
+    FlushThreadTaskRunner(std::move(task_runner));
+  }
 }
 
 NOINLINE void ResetThread_CACHE(
-    std::unique_ptr<BrowserProcessSubThread> thread) {
+    std::unique_ptr<BrowserProcessSubThread> thread,
+    scoped_refptr<base::SingleThreadTaskRunner> task_runner) {
+  DCHECK((thread && !task_runner) || (!thread && task_runner));
   volatile int inhibit_comdat = __LINE__;
   ALLOW_UNUSED_LOCAL(inhibit_comdat);
-  thread.reset();
+  if (thread) {
+    thread.reset();
+  } else {
+    BrowserThreadImpl::RedirectThreadIDToTaskRunner(BrowserThread::CACHE,
+                                                    nullptr);
+    FlushThreadTaskRunner(std::move(task_runner));
+  }
 }
 
 NOINLINE void ResetThread_IO(std::unique_ptr<BrowserProcessSubThread> thread) {
   volatile int inhibit_comdat = __LINE__;
   ALLOW_UNUSED_LOCAL(inhibit_comdat);
   thread.reset();
 }
 
 NOINLINE void ResetThread_IndexedDb(std::unique_ptr<base::Thread> thread) {
   volatile int inhibit_comdat = __LINE__;
@@ skipping 530 matching lines @@
   ui_message_loop_options.message_loop_type = base::MessageLoop::TYPE_UI;
 
   // Start threads in the order they occur in the BrowserThread::ID
   // enumeration, except for BrowserThread::UI which is the main
   // thread.
   //
   // Must be size_t so we can increment it.
   for (size_t thread_id = BrowserThread::UI + 1;
        thread_id < BrowserThread::ID_COUNT;
        ++thread_id) {
-    std::unique_ptr<BrowserProcessSubThread>* thread_to_start = NULL;
+    // If this thread ID is backed by a real thread, |thread_to_start| will be
+    // set to the appropriate |BrowserProcessSubThread*|. And |options| can be

[fdoray, 2016/12/01 15:15:21]

s/|BrowserProcessSubThread*|/BrowserProcessSubThread*

[gab, 2016/12/07 19:15:26]

Done.

+    // updated away from its default.
+    std::unique_ptr<BrowserProcessSubThread>* thread_to_start = nullptr;
     base::Thread::Options options;
 
+    // Otherwise this thread ID will be backed by a SingleThreadTaskRunner in
+    // which case |task_runner_to_create| will be set to the appropriate
+    // |SingleThreadTaskRunner*| and |task_runner_traits| can be updated away

[fdoray, 2016/12/01 15:15:21]

s/|SingleThreadTaskRunner*|/SingleThreadTaskRunner*

[gab, 2016/12/07 19:15:26]

Done.

+    // from its default. |execution_mode| defaults to |SINGLE_THREADED| as-is

[fdoray, 2016/12/01 15:15:21]

No more |execution_mode|.

[gab, 2016/12/07 19:15:26]

Done.

+    // appropriate to map most old browser threads but can also be updated in
+    // the switch statement below if a thread has specific requirements.
+    scoped_refptr<base::SingleThreadTaskRunner>* task_runner_to_create =
+        nullptr;
+    base::TaskTraits task_runner_traits;
+
+    const bool redirect_nonUInonIO_browser_threads =

[fdoray, 2016/12/01 15:15:21]

Move outside the loop.

[gab, 2016/12/07 19:15:27]

Done.

+        GetContentClient()
+            ->browser()
+            ->RedirectNonUINonIOBrowserThreadsToTaskScheduler();
+
     switch (thread_id) {
       case BrowserThread::DB:
         TRACE_EVENT_BEGIN1("startup",
             "BrowserMainLoop::CreateThreads:start",
             "Thread", "BrowserThread::DB");
-        thread_to_start = &db_thread_;
-        options.timer_slack = base::TIMER_SLACK_MAXIMUM;
+        if (redirect_nonUInonIO_browser_threads) {
+          task_runner_to_create = &db_task_runner_;
+          task_runner_traits.WithFileIO()

[fdoray, 2016/12/01 15:15:21]

Add WithWait() to allow AssertIOAllowed()?

[gab, 2016/12/07 19:15:27]

Assuming you mean "to allow AssertWaitAllowed()". Looks like only UI/IO threads
DisallowWaiting() at the moment so I'll add WithWait() to the other ones for now
with a TODO to look into removing it.

+              .WithPriority(base::TaskPriority::USER_VISIBLE)
+              .WithShutdownBehavior(base::TaskShutdownBehavior::BLOCK_SHUTDOWN);
+        } else {
+          thread_to_start = &db_thread_;
+          options.timer_slack = base::TIMER_SLACK_MAXIMUM;
+        }
         break;
       case BrowserThread::FILE_USER_BLOCKING:
         TRACE_EVENT_BEGIN1("startup",
             "BrowserMainLoop::CreateThreads:start",
             "Thread", "BrowserThread::FILE_USER_BLOCKING");
-        thread_to_start = &file_user_blocking_thread_;
+        if (redirect_nonUInonIO_browser_threads) {
+          task_runner_to_create = &file_user_blocking_task_runner_;
+          task_runner_traits.WithFileIO()
+              .WithPriority(base::TaskPriority::USER_BLOCKING)
+              .WithShutdownBehavior(base::TaskShutdownBehavior::BLOCK_SHUTDOWN);
+        } else {
+          thread_to_start = &file_user_blocking_thread_;
+        }
         break;
       case BrowserThread::FILE:
         TRACE_EVENT_BEGIN1("startup",
             "BrowserMainLoop::CreateThreads:start",
             "Thread", "BrowserThread::FILE");
+
+#if defined(OS_WIN)
+        // On Windows, the FILE thread needs to be have a UI message loop which
+        // pumps messages in such a way that Google Update can communicate back
+        // to us.
+        // TODO(robliao): Need to support COM in TaskScheduler before
+        // redirecting the FILE thread on Windows. http://crbug.com/662122
         thread_to_start = &file_thread_;
-#if defined(OS_WIN)
-        // On Windows, the FILE thread needs to be have a UI message loop
-        // which pumps messages in such a way that Google Update can
-        // communicate back to us.
         options = ui_message_loop_options;
 #else
-        options = io_message_loop_options;
+        if (redirect_nonUInonIO_browser_threads) {
+          task_runner_to_create = &file_task_runner_;
+          task_runner_traits.WithFileIO()
+              .WithPriority(base::TaskPriority::BACKGROUND)
+              .WithShutdownBehavior(base::TaskShutdownBehavior::BLOCK_SHUTDOWN);
+        } else {
+          thread_to_start = &file_thread_;
+          options = io_message_loop_options;
+          options.timer_slack = base::TIMER_SLACK_MAXIMUM;
+        }
 #endif
-        options.timer_slack = base::TIMER_SLACK_MAXIMUM;
         break;
       case BrowserThread::PROCESS_LAUNCHER:
         TRACE_EVENT_BEGIN1("startup",
             "BrowserMainLoop::CreateThreads:start",
             "Thread", "BrowserThread::PROCESS_LAUNCHER");
-        thread_to_start = &process_launcher_thread_;
-        options.timer_slack = base::TIMER_SLACK_MAXIMUM;
+        if (redirect_nonUInonIO_browser_threads) {
+          task_runner_to_create = &process_launcher_task_runner_;
+          task_runner_traits.WithFileIO()
+              .WithPriority(base::TaskPriority::USER_BLOCKING)
+              .WithShutdownBehavior(base::TaskShutdownBehavior::BLOCK_SHUTDOWN);
+        } else {
+          thread_to_start = &process_launcher_thread_;
+          options.timer_slack = base::TIMER_SLACK_MAXIMUM;
+        }
         break;
       case BrowserThread::CACHE:
         TRACE_EVENT_BEGIN1("startup",
             "BrowserMainLoop::CreateThreads:start",
             "Thread", "BrowserThread::CACHE");
+#if defined(OS_WIN)
+        // TaskScheduler doesn't support async I/O on Windows as CACHE thread is
+        // the only user and this use case is going away in
+        // https://codereview.chromium.org/2216583003/. TODO(gavinp): Remove
+        // this ifdef (and thus enable redirection of the CACHE thread on
+        // Windows) once that CL lands.
         thread_to_start = &cache_thread_;
-#if defined(OS_WIN)
         options = io_message_loop_options;
-#endif  // defined(OS_WIN)
         options.timer_slack = base::TIMER_SLACK_MAXIMUM;
+#else  // OS_WIN
+        if (redirect_nonUInonIO_browser_threads) {
+          task_runner_to_create = &cache_task_runner_;
+          task_runner_traits.WithFileIO()
+              .WithPriority(base::TaskPriority::USER_BLOCKING)
+              .WithShutdownBehavior(base::TaskShutdownBehavior::BLOCK_SHUTDOWN);
+        } else {
+          thread_to_start = &cache_thread_;
+          options.timer_slack = base::TIMER_SLACK_MAXIMUM;
+        }
+#endif  // OS_WIN
         break;
       case BrowserThread::IO:
         TRACE_EVENT_BEGIN1("startup",
             "BrowserMainLoop::CreateThreads:start",
             "Thread", "BrowserThread::IO");
         thread_to_start = &io_thread_;
         options = io_message_loop_options;
 #if defined(OS_ANDROID) || defined(OS_CHROMEOS)
         // Up the priority of the |io_thread_| as some of its IPCs relate to
         // display tasks.
         options.priority = base::ThreadPriority::DISPLAY;
 #endif
         break;
-      case BrowserThread::UI:
-      case BrowserThread::ID_COUNT:
-      default:
+      case BrowserThread::UI:        // Falls through.
+      case BrowserThread::ID_COUNT:  // Falls through.
         NOTREACHED();
         break;
     }
 
     BrowserThread::ID id = static_cast<BrowserThread::ID>(thread_id);
 
     if (thread_to_start) {
+      DCHECK(!task_runner_to_create);
       (*thread_to_start).reset(new BrowserProcessSubThread(id));
       if (!(*thread_to_start)->StartWithOptions(options)) {
         LOG(FATAL) << "Failed to start the browser thread: id == " << id;
       }
+    } else if (task_runner_to_create) {

[fdoray, 2016/12/01 15:15:21]

DCHECK(!thread_to_start);

[gab, 2016/12/07 19:15:27]

This is implicit in else block of if (thread_to_start)

+      *task_runner_to_create =
+          base::CreateSingleThreadTaskRunnerWithTraits(task_runner_traits);
+      DCHECK(*task_runner_to_create);
+      BrowserThreadImpl::RedirectThreadIDToTaskRunner(
+          id, task_runner_to_create->get());
     } else {
       NOTREACHED();
     }
 
     TRACE_EVENT_END0("startup", "BrowserMainLoop::CreateThreads:start");
   }
   created_threads_ = true;
   return result_code_;
 }
 
@@ skipping 132 matching lines @@
     // - The IO thread is the only user of the CACHE thread.
     //
     // - The PROCESS_LAUNCHER thread must be stopped after IO in case
     //   the IO thread posted a task to terminate a process on the
     //   process launcher thread.
     //
     // - (Not sure why DB stops last.)
     switch (thread_id) {
       case BrowserThread::DB: {
         TRACE_EVENT0("shutdown", "BrowserMainLoop::Subsystem:DBThread");
-        ResetThread_DB(std::move(db_thread_));
+        ResetThread_DB(std::move(db_thread_), std::move(db_task_runner_));
         break;
       }
       case BrowserThread::FILE: {
         TRACE_EVENT0("shutdown", "BrowserMainLoop::Subsystem:FileThread");
-        // Clean up state that lives on or uses the file_thread_ before
-        // it goes away.
+        // Clean up state that lives on or uses the file_thread_ before it goes

[fdoray, 2016/12/01 15:15:21]

the FILE thread

[gab, 2016/12/07 19:15:26]

Done.

+        // away.
         save_file_manager_->Shutdown();
-        ResetThread_FILE(std::move(file_thread_));
+        ResetThread_FILE(std::move(file_thread_), std::move(file_task_runner_));
         break;
       }
       case BrowserThread::FILE_USER_BLOCKING: {
         TRACE_EVENT0("shutdown",
                       "BrowserMainLoop::Subsystem:FileUserBlockingThread");
-        ResetThread_FILE_USER_BLOCKING(std::move(file_user_blocking_thread_));
+        ResetThread_FILE_USER_BLOCKING(
+            std::move(file_user_blocking_thread_),
+            std::move(file_user_blocking_task_runner_));
         break;
       }
       case BrowserThread::PROCESS_LAUNCHER: {
         TRACE_EVENT0("shutdown", "BrowserMainLoop::Subsystem:LauncherThread");
-        ResetThread_PROCESS_LAUNCHER(std::move(process_launcher_thread_));
+        ResetThread_PROCESS_LAUNCHER(std::move(process_launcher_thread_),
+                                     std::move(process_launcher_task_runner_));
         break;
       }
       case BrowserThread::CACHE: {
         TRACE_EVENT0("shutdown", "BrowserMainLoop::Subsystem:CacheThread");
-        ResetThread_CACHE(std::move(cache_thread_));
+        ResetThread_CACHE(std::move(cache_thread_),
+                          std::move(cache_task_runner_));
         break;
       }
       case BrowserThread::IO: {
         TRACE_EVENT0("shutdown", "BrowserMainLoop::Subsystem:IOThread");
         ResetThread_IO(std::move(io_thread_));
         break;
       }
       case BrowserThread::UI:
       case BrowserThread::ID_COUNT:
-      default:
         NOTREACHED();
         break;
     }
   }
   {
     TRACE_EVENT0("shutdown", "BrowserMainLoop::Subsystem:IndexedDBThread");
     ResetThread_IndexedDb(std::move(indexed_db_thread_));
   }
 
   // Close the blocking I/O pool after the other threads. Other threads such
@@ skipping 468 matching lines @@
     scoped_refptr<base::SingleThreadTaskRunner> worker_task_runner =
         audio_thread_->task_runner();
     audio_manager_ = media::AudioManager::Create(std::move(audio_task_runner),
                                                  std::move(worker_task_runner),
                                                  MediaInternals::GetInstance());
   }
   CHECK(audio_manager_);
 }
 
 }  // namespace content