[Chromoting] Introducing refcount-based life time management of the message loops in the service (daemon) and me2me host (network) processes.

This CL introduces AutoMessageLoop wrapper that provides control over life time of a message loop via scoped_refptr references. This scheme is useful in the cases when shutdown code has to run on a particular thread or when the OS requires resources (such as windows) to be freed before exiting a message loop.

The CL switches threads, owned by remoting::HostService, remoting::HostProcess and remoting::ChromotingHostContext, to refcount-based lifetime management. This change required updating tear-down sequences in remoting_me2me_host and the host plugin code.

BUG=134694


Review URL: https://chromiumcodereview.appspot.com/10829467

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@154827 0039d316-1c4b-4281-b951-d872f2087c98

12 files changed, 233 insertions, 117 deletions

diff --git a/remoting/host/chromoting_host_context.cc b/remoting/host/chromoting_host_context.cc
index b110567..3aa26af 100644
--- a/remoting/host/chromoting_host_context.cc
+++ b/remoting/host/chromoting_host_context.cc
@@ -8,24 +8,36 @@
 
 #include "base/bind.h"
 #include "base/threading/thread.h"
+#include "remoting/base/auto_thread_task_runner.h"
 #include "remoting/host/url_request_context.h"
 
 namespace remoting {
 
 ChromotingHostContext::ChromotingHostContext(
-    scoped_refptr<base::SingleThreadTaskRunner> ui_task_runner)
-    : network_thread_("ChromotingNetworkThread"),
+    scoped_refptr<AutoThreadTaskRunner> ui_task_runner)
+    : audio_thread_("ChromotingAudioThread"),
       capture_thread_("ChromotingCaptureThread"),
-      encode_thread_("ChromotingEncodeThread"),
-      audio_thread_("ChromotingAudioThread"),
       desktop_thread_("ChromotingDesktopThread"),
+      encode_thread_("ChromotingEncodeThread"),
       file_thread_("ChromotingFileIOThread"),
+      network_thread_("ChromotingNetworkThread"),
       ui_task_runner_(ui_task_runner) {
 }
 
 ChromotingHostContext::~ChromotingHostContext() {
 }
 
+void ChromotingHostContext::ReleaseTaskRunners() {
+  url_request_context_getter_ = NULL;
+  audio_task_runner_ = NULL;
+  capture_task_runner_ = NULL;
+  desktop_task_runner_ = NULL;
+  encode_task_runner_ = NULL;
+  file_task_runner_ = NULL;
+  network_task_runner_ = NULL;
+  ui_task_runner_ = NULL;
+}
+
 bool ChromotingHostContext::Start() {
   // Start all the threads.
   bool started = capture_thread_.Start() && encode_thread_.Start() &&
@@ -39,40 +51,62 @@ bool ChromotingHostContext::Start() {
   if (!started)
     return false;
 
+  // Wrap worker threads with |AutoThreadTaskRunner| and have them reference
+  // the main thread via |ui_task_runner_|, to ensure that it remain active to
+  // Stop() them when no references remain.
+  audio_task_runner_ =
+      new AutoThreadTaskRunner(audio_thread_.message_loop_proxy(),
+                               ui_task_runner_);
+  capture_task_runner_ =
+      new AutoThreadTaskRunner(capture_thread_.message_loop_proxy(),
+                               ui_task_runner_);
+  desktop_task_runner_ =
+      new AutoThreadTaskRunner(desktop_thread_.message_loop_proxy(),
+                               ui_task_runner_);
+  encode_task_runner_ =
+      new AutoThreadTaskRunner(encode_thread_.message_loop_proxy(),
+                               ui_task_runner_);
+  file_task_runner_ =
+      new AutoThreadTaskRunner(file_thread_.message_loop_proxy(),
+                               ui_task_runner_);
+
+  network_task_runner_ =
+      new AutoThreadTaskRunner(network_thread_.message_loop_proxy(),
+                               ui_task_runner_);
   url_request_context_getter_ = new URLRequestContextGetter(
       ui_task_runner(), network_task_runner(),
       static_cast<MessageLoopForIO*>(file_thread_.message_loop()));
   return true;
 }
 
+base::SingleThreadTaskRunner* ChromotingHostContext::audio_task_runner() {
+  return audio_task_runner_;
+}
+
 base::SingleThreadTaskRunner* ChromotingHostContext::capture_task_runner() {
-  return capture_thread_.message_loop_proxy();
+  return capture_task_runner_;
 }
 
-base::SingleThreadTaskRunner* ChromotingHostContext::encode_task_runner() {
-  return encode_thread_.message_loop_proxy();
+base::SingleThreadTaskRunner* ChromotingHostContext::desktop_task_runner() {
+  return desktop_task_runner_;
 }
 
-base::SingleThreadTaskRunner* ChromotingHostContext::audio_task_runner() {
-  return audio_thread_.message_loop_proxy();
+base::SingleThreadTaskRunner* ChromotingHostContext::encode_task_runner() {
+  return encode_task_runner_;
 }
 
-base::SingleThreadTaskRunner* ChromotingHostContext::network_task_runner() {
-  return network_thread_.message_loop_proxy();
+base::SingleThreadTaskRunner* ChromotingHostContext::file_task_runner() {
+  return file_task_runner_;
 }
 
-base::SingleThreadTaskRunner* ChromotingHostContext::desktop_task_runner() {
-  return desktop_thread_.message_loop_proxy();
+base::SingleThreadTaskRunner* ChromotingHostContext::network_task_runner() {
+  return network_task_runner_;
 }
 
 base::SingleThreadTaskRunner* ChromotingHostContext::ui_task_runner() {
   return ui_task_runner_;
 }
 
-base::SingleThreadTaskRunner* ChromotingHostContext::file_task_runner() {
-  return file_thread_.message_loop_proxy();
-}
-
 const scoped_refptr<net::URLRequestContextGetter>&
 ChromotingHostContext::url_request_context_getter() {
   DCHECK(url_request_context_getter_.get());
diff --git a/remoting/host/chromoting_host_context.h b/remoting/host/chromoting_host_context.h
index 149444a..497b8e3 100644
--- a/remoting/host/chromoting_host_context.h
+++ b/remoting/host/chromoting_host_context.h
@@ -5,22 +5,17 @@
 #ifndef REMOTING_HOST_CHROMOTING_HOST_CONTEXT_H_
 #define REMOTING_HOST_CHROMOTING_HOST_CONTEXT_H_
 
-#include <string>
-
 #include "base/gtest_prod_util.h"
 #include "base/memory/ref_counted.h"
-#include "base/threading/platform_thread.h"
+#include "base/single_thread_task_runner.h"
 #include "base/threading/thread.h"
 
-namespace base {
-class SingleThreadTaskRunner;
-}  // namespace base
-
 namespace net {
 class URLRequestContextGetter;
 }  // namespace net
 
 namespace remoting {
+class AutoThreadTaskRunner;
 
 // A class that manages threads and running context for the chromoting host
 // process.  This class is virtual only for testing purposes (see below).
@@ -28,9 +23,11 @@ class ChromotingHostContext {
  public:
   // Create a context.
   ChromotingHostContext(
-      scoped_refptr<base::SingleThreadTaskRunner> ui_task_runner);
+      scoped_refptr<AutoThreadTaskRunner> ui_task_runner);
   virtual ~ChromotingHostContext();
 
+  void ReleaseTaskRunners();
+
   // TODO(ajwong): Move the Start method out of this class. Then
   // create a static factory for construction, and destruction.  We
   // should be able to remove the need for virtual functions below
@@ -38,61 +35,70 @@ class ChromotingHostContext {
   // this API.
   virtual bool Start();
 
-  // Task runner for the thread that is used for the UI. In the NPAPI
-  // plugin this corresponds to the main plugin thread.
-  virtual base::SingleThreadTaskRunner* ui_task_runner();
+  // Task runner for the thread used for audio capture and encoding.
+  virtual base::SingleThreadTaskRunner* audio_task_runner();
 
   // Task runner for the thread used by the ScreenRecorder to capture
   // the screen.
   virtual base::SingleThreadTaskRunner* capture_task_runner();
 
-  // Task runner for the thread used to encode video streams.
-  virtual base::SingleThreadTaskRunner* encode_task_runner();
-
-  // Task runner for the thread used for audio capture and encoding.
-  virtual base::SingleThreadTaskRunner* audio_task_runner();
-
-  // Task runner for the thread used for network IO. This thread runs
-  // a libjingle message loop, and is the only thread on which
-  // libjingle code may be run.
-  virtual base::SingleThreadTaskRunner* network_task_runner();
-
   // Task runner for the thread that is used by the EventExecutor.
   //
   // TODO(sergeyu): Do we need a separate thread for EventExecutor?
   // Can we use some other thread instead?
   virtual base::SingleThreadTaskRunner* desktop_task_runner();
 
+  // Task runner for the thread used to encode video streams.
+  virtual base::SingleThreadTaskRunner* encode_task_runner();
+
   // Task runner for the thread that is used for blocking file
   // IO. This thread is used by the URLRequestContext to read proxy
   // configuration and by NatConfig to read policy configs.
   virtual base::SingleThreadTaskRunner* file_task_runner();
 
+  // Task runner for the thread used for network IO. This thread runs
+  // a libjingle message loop, and is the only thread on which
+  // libjingle code may be run.
+  virtual base::SingleThreadTaskRunner* network_task_runner();
+
+  // Task runner for the thread that is used for the UI. In the NPAPI
+  // plugin this corresponds to the main plugin thread.
+  virtual base::SingleThreadTaskRunner* ui_task_runner();
+
   const scoped_refptr<net::URLRequestContextGetter>&
       url_request_context_getter();
 
  private:
   FRIEND_TEST_ALL_PREFIXES(ChromotingHostContextTest, StartAndStop);
 
-  // A thread that hosts all network operations.
-  base::Thread network_thread_;
+  // A thread that hosts audio capture and encoding.
+  base::Thread audio_thread_;
 
   // A thread that hosts screen capture.
   base::Thread capture_thread_;
 
-  // A thread that hosts all encode operations.
-  base::Thread encode_thread_;
-
-  // A thread that hosts audio capture and encoding.
-  base::Thread audio_thread_;
-
   // A thread that hosts input injection.
   base::Thread desktop_thread_;
 
+  // A thread that hosts all encode operations.
+  base::Thread encode_thread_;
+
   // Thread for blocking IO operations.
   base::Thread file_thread_;
 
-  scoped_refptr<base::SingleThreadTaskRunner> ui_task_runner_;
+  // A thread that hosts all network operations.
+  base::Thread network_thread_;
+
+  // Task runners wrapping the above threads. These should be declared after
+  // the corresponding threads to guarantee proper order of destruction.
+  scoped_refptr<base::SingleThreadTaskRunner> audio_task_runner_;
+  scoped_refptr<base::SingleThreadTaskRunner> capture_task_runner_;
+  scoped_refptr<base::SingleThreadTaskRunner> desktop_task_runner_;
+  scoped_refptr<base::SingleThreadTaskRunner> encode_task_runner_;
+  scoped_refptr<base::SingleThreadTaskRunner> file_task_runner_;
+  scoped_refptr<base::SingleThreadTaskRunner> network_task_runner_;
+
+  scoped_refptr<AutoThreadTaskRunner> ui_task_runner_;
 
   scoped_refptr<net::URLRequestContextGetter> url_request_context_getter_;
 
diff --git a/remoting/host/chromoting_host_context_unittest.cc b/remoting/host/chromoting_host_context_unittest.cc
index 1ceb95c..a4e670f 100644
--- a/remoting/host/chromoting_host_context_unittest.cc
+++ b/remoting/host/chromoting_host_context_unittest.cc
@@ -4,6 +4,7 @@
 
 #include "base/message_loop.h"
 #include "base/message_loop_proxy.h"
+#include "remoting/base/auto_thread_task_runner.h"
 #include "remoting/host/chromoting_host_context.h"
 #include "testing/gtest/include/gtest/gtest.h"
 
@@ -13,7 +14,8 @@ namespace remoting {
 // operates properly and all threads and message loops are valid.
 TEST(ChromotingHostContextTest, StartAndStop) {
   MessageLoopForUI message_loop;
-  ChromotingHostContext context(base::MessageLoopProxy::current());
+  ChromotingHostContext context(new AutoThreadTaskRunner(
+      base::MessageLoopProxy::current()));
 
   context.Start();
   EXPECT_TRUE(context.network_task_runner());
diff --git a/remoting/host/chromoting_host_unittest.cc b/remoting/host/chromoting_host_unittest.cc
index 2fbf2cc..3a82f0c 100644
--- a/remoting/host/chromoting_host_unittest.cc
+++ b/remoting/host/chromoting_host_unittest.cc
@@ -368,6 +368,7 @@ class ChromotingHostTest : public testing::Test {
  protected:
   MessageLoop message_loop_;
   scoped_refptr<base::MessageLoopProxy> message_loop_proxy_;
+  MockChromotingHostContext context_;
   MockConnectionToClientEventHandler handler_;
   MockSignalStrategy signal_strategy_;
   MockEventExecutor* event_executor_;
@@ -375,7 +376,6 @@ class ChromotingHostTest : public testing::Test {
   scoped_ptr<It2MeHostUserInterface> it2me_host_user_interface_;
   scoped_refptr<ChromotingHost> host_;
   MockHostStatusObserver host_status_observer_;
-  MockChromotingHostContext context_;
   protocol::MockSessionManager* session_manager_;
   std::string xmpp_login_;
   MockConnectionToClient* connection1_;
diff --git a/remoting/host/client_session_unittest.cc b/remoting/host/client_session_unittest.cc
index 84783bb..bf54230 100644
--- a/remoting/host/client_session_unittest.cc
+++ b/remoting/host/client_session_unittest.cc
@@ -2,6 +2,7 @@
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
+#include "base/message_loop.h"
 #include "remoting/base/constants.h"
 #include "remoting/host/client_session.h"
 #include "remoting/host/host_mock_objects.h"
diff --git a/remoting/host/constants.h b/remoting/host/constants.h
index 0d37c38..e2723b6 100644
--- a/remoting/host/constants.h
+++ b/remoting/host/constants.h
@@ -18,6 +18,7 @@ enum HostExitCodes {
   // Error codes that don't indicate a permanent error condition.
   kSuccessExitCode = 0,
   kReservedForX11ExitCode = 1,
+  kHostInitializationFailed = 7,
 
   // Error codes that do indicate a permanent error condition.
   kInvalidHostConfigurationExitCode = 2,
diff --git a/remoting/host/host_mock_objects.cc b/remoting/host/host_mock_objects.cc
index f192f8e..f8867a3 100644
--- a/remoting/host/host_mock_objects.cc
+++ b/remoting/host/host_mock_objects.cc
@@ -6,6 +6,7 @@
 
 #include "base/message_loop_proxy.h"
 #include "net/base/ip_endpoint.h"
+#include "remoting/base/auto_thread_task_runner.h"
 #include "remoting/base/capture_data.h"
 #include "remoting/proto/event.pb.h"
 #include "remoting/protocol/transport.h"
@@ -59,7 +60,8 @@ scoped_ptr<LocalInputMonitor> LocalInputMonitor::Create() {
 }
 
 MockChromotingHostContext::MockChromotingHostContext()
-    : ChromotingHostContext(base::MessageLoopProxy::current()) {
+    : ChromotingHostContext(new AutoThreadTaskRunner(
+          base::MessageLoopProxy::current())) {
 }
 
 MockChromotingHostContext::~MockChromotingHostContext() {}
diff --git a/remoting/host/plugin/host_script_object.cc b/remoting/host/plugin/host_script_object.cc
index 118114f..8cffb5e 100644
--- a/remoting/host/plugin/host_script_object.cc
+++ b/remoting/host/plugin/host_script_object.cc
@@ -15,6 +15,7 @@
 #include "base/utf_string_conversions.h"
 #include "base/values.h"
 #include "net/base/net_util.h"
+#include "remoting/base/auto_thread_task_runner.h"
 #include "remoting/base/auth_token_util.h"
 #include "remoting/host/chromoting_host.h"
 #include "remoting/host/chromoting_host_context.h"
@@ -106,6 +107,10 @@ HostNPScriptObject::~HostNPScriptObject() {
 
   HostLogHandler::UnregisterLoggingScriptObject(this);
 
+  // Stop the message loop. Any attempt to post a task to
+  // |context_.ui_task_runner()| will result in a CHECK() after this point.
+  // TODO(alexeypa): Enable posting messages to |plugin_task_runner_| during
+  // shutdown to avoid this hack.
   plugin_task_runner_->Detach();
 
   // Stop listening for policy updates.
@@ -121,9 +126,7 @@ HostNPScriptObject::~HostNPScriptObject() {
     // here because |host_context_| needs to be stopped on the plugin
     // thread, but the plugin thread may not exist after the instance
     // is destroyed.
-    disconnected_event_.Reset();
     DisconnectInternal();
-    disconnected_event_.Wait();
 
     // UI needs to be shut down on the UI thread before we destroy the
     // host context (because it depends on the context object), but
@@ -132,7 +135,15 @@ HostNPScriptObject::~HostNPScriptObject() {
     // unregister it from this thread).
     it2me_host_user_interface_.reset();
 
-    // Stops all threads.
+    // Release the context's TaskRunner references for the threads, so they can
+    // exit when no objects need them.
+    host_context_->ReleaseTaskRunners();
+
+    // |disconnected_event_| is signalled when the last reference to the plugin
+    // thread is dropped.
+    disconnected_event_.Wait();
+
+    // Stop all threads.
     host_context_.reset();
   }
 
@@ -143,7 +154,10 @@ bool HostNPScriptObject::Init() {
   DCHECK(plugin_task_runner_->BelongsToCurrentThread());
   VLOG(2) << "Init";
 
-  host_context_.reset(new ChromotingHostContext(plugin_task_runner_));
+  host_context_.reset(new ChromotingHostContext(new AutoThreadTaskRunner(
+      plugin_task_runner_,
+      base::Bind(&base::WaitableEvent::Signal,
+                 base::Unretained(&disconnected_event_)))));
   if (!host_context_->Start()) {
     host_context_.reset();
     return false;
@@ -878,13 +892,12 @@ void HostNPScriptObject::DisconnectInternal() {
 
   switch (state_) {
     case kDisconnected:
-      disconnected_event_.Signal();
       return;
 
     case kStarting:
+      desktop_environment_.reset();
       SetState(kDisconnecting);
       SetState(kDisconnected);
-      disconnected_event_.Signal();
       return;
 
     case kDisconnecting:
@@ -913,7 +926,7 @@ void HostNPScriptObject::DisconnectInternal() {
 void HostNPScriptObject::OnShutdownFinished() {
   DCHECK(host_context_->network_task_runner()->BelongsToCurrentThread());
 
-  disconnected_event_.Signal();
+  desktop_environment_.reset();
 }
 
 void HostNPScriptObject::OnPolicyUpdate(
diff --git a/remoting/host/remoting_me2me_host.cc b/remoting/host/remoting_me2me_host.cc
index 2d67463..192cfe2 100644
--- a/remoting/host/remoting_me2me_host.cc
+++ b/remoting/host/remoting_me2me_host.cc
@@ -28,6 +28,7 @@
 #include "ipc/ipc_channel_proxy.h"
 #include "net/base/network_change_notifier.h"
 #include "net/socket/ssl_server_socket.h"
+#include "remoting/base/auto_thread_task_runner.h"
 #include "remoting/base/breakpad.h"
 #include "remoting/base/constants.h"
 #include "remoting/host/branding.h"
@@ -95,6 +96,10 @@ const char kOfficialOAuth2ClientId[] =
     "440925447803-avn2sj1kc099s0r7v62je5s339mu0am1.apps.googleusercontent.com";
 const char kOfficialOAuth2ClientSecret[] = "Bgur6DFiOMM1h8x-AQpuTQlK";
 
+void QuitMessageLoop(MessageLoop* message_loop) {
+  message_loop->PostTask(FROM_HERE, MessageLoop::QuitClosure());
+}
+
 }  // namespace
 
 namespace remoting {
@@ -103,8 +108,8 @@ class HostProcess
     : public HeartbeatSender::Listener,
       public IPC::Listener {
  public:
-  HostProcess()
-      : message_loop_(MessageLoop::TYPE_UI),
+  HostProcess(scoped_ptr<ChromotingHostContext> context)
+      : context_(context.Pass()),
 #ifdef OFFICIAL_BUILD
         oauth_use_official_client_id_(true),
 #else
@@ -121,9 +126,6 @@ class HostProcess
                             base::Unretained(this)))
 #endif
   {
-    context_.reset(
-        new ChromotingHostContext(message_loop_.message_loop_proxy()));
-    context_->Start();
     network_change_notifier_.reset(net::NetworkChangeNotifier::Create());
     config_updated_timer_.reset(new base::DelayTimer<HostProcess>(
         FROM_HERE, base::TimeDelta::FromSeconds(2), this,
@@ -162,7 +164,7 @@ class HostProcess
   }
 
   void ConfigUpdated() {
-    DCHECK(message_loop_.message_loop_proxy()->BelongsToCurrentThread());
+    DCHECK(context_->ui_task_runner()->BelongsToCurrentThread());
 
     // Call ConfigUpdatedDelayed after a short delay, so that this object won't
     // try to read the updated configuration file before it has been
@@ -173,7 +175,7 @@ class HostProcess
   }
 
   void ConfigUpdatedDelayed() {
-    DCHECK(message_loop_.message_loop_proxy()->BelongsToCurrentThread());
+    DCHECK(context_->ui_task_runner()->BelongsToCurrentThread());
 
     if (LoadConfig()) {
       // PostTask to create new authenticator factory in case PIN has changed.
@@ -218,7 +220,7 @@ class HostProcess
 #elif defined(OS_WIN)
     scoped_refptr<base::files::FilePathWatcher::Delegate> delegate(
         new ConfigChangedDelegate(
-            message_loop_.message_loop_proxy(),
+            context_->ui_task_runner(),
             base::Bind(&HostProcess::ConfigUpdated, base::Unretained(this))));
     config_watcher_.reset(new base::files::FilePathWatcher());
     if (!config_watcher_->Watch(host_config_path_, delegate)) {
@@ -241,9 +243,16 @@ class HostProcess
     return false;
   }
 
-  int Run() {
-    if (!LoadConfig()) {
-      return kInvalidHostConfigurationExitCode;
+  void StartHostProcess() {
+    DCHECK(context_->ui_task_runner()->BelongsToCurrentThread());
+
+    if (!InitWithCommandLine(CommandLine::ForCurrentProcess()) ||
+        !LoadConfig()) {
+      context_->network_task_runner()->PostTask(
+          FROM_HERE,
+          base::Bind(&HostProcess::Shutdown, base::Unretained(this),
+                     kInvalidHostConfigurationExitCode));
+      return;
     }
 
 #if defined(OS_MACOSX) || defined(OS_WIN)
@@ -275,19 +284,28 @@ class HostProcess
         base::Bind(&HostProcess::ListenForConfigChanges,
                    base::Unretained(this)));
 #endif
-    message_loop_.Run();
+  }
+
+  int get_exit_code() const { return exit_code_; }
+
+ private:
+  void ShutdownHostProcess() {
+    DCHECK(context_->ui_task_runner()->BelongsToCurrentThread());
+
+    daemon_channel_.reset();
 
 #if defined(OS_MACOSX) || defined(OS_WIN)
     host_user_interface_.reset();
 #endif
 
-    daemon_channel_.reset();
-    base::WaitableEvent done_event(true, false);
-    policy_watcher_->StopWatching(&done_event);
-    done_event.Wait();
-    policy_watcher_.reset();
+    if (policy_watcher_.get()) {
+      base::WaitableEvent done_event(true, false);
+      policy_watcher_->StopWatching(&done_event);
+      done_event.Wait();
+      policy_watcher_.reset();
+    }
 
-    return exit_code_;
+    context_.reset();
   }
 
   // Overridden from HeartbeatSender::Listener
@@ -296,7 +314,6 @@ class HostProcess
     Shutdown(kInvalidHostIdExitCode);
   }
 
- private:
   void StartWatchingPolicy() {
     policy_watcher_.reset(
         policy_hack::PolicyWatcher::Create(context_->file_task_runner()));
@@ -306,7 +323,7 @@ class HostProcess
 
   // Read host config, returning true if successful.
   bool LoadConfig() {
-    DCHECK(message_loop_.message_loop_proxy()->BelongsToCurrentThread());
+    DCHECK(context_->ui_task_runner()->BelongsToCurrentThread());
 
     // TODO(sergeyu): There is a potential race condition: this function is
     // called on the main thread while the class members it mutates are used on
@@ -575,7 +592,7 @@ class HostProcess
   // Invoked when the user uses the Disconnect windows to terminate
   // the sessions.
   void OnDisconnectRequested() {
-    DCHECK(message_loop_.message_loop_proxy()->BelongsToCurrentThread());
+    DCHECK(context_->ui_task_runner()->BelongsToCurrentThread());
 
     host_->DisconnectAllClients();
   }
@@ -627,12 +644,17 @@ class HostProcess
     host_ = NULL;
     ResetHost();
 
-    message_loop_.PostTask(FROM_HERE, MessageLoop::QuitClosure());
+    // Complete the rest of shutdown on the main thread.
+    context_->ui_task_runner()->PostTask(
+        FROM_HERE,
+        base::Bind(&HostProcess::ShutdownHostProcess,
+                   base::Unretained(this)));
   }
 
   void ResetHost() {
     DCHECK(context_->network_task_runner()->BelongsToCurrentThread());
 
+    desktop_environment_.reset();
     host_event_logger_.reset();
     log_to_server_.reset();
     heartbeat_sender_.reset();
@@ -640,7 +662,6 @@ class HostProcess
     signal_strategy_.reset();
   }
 
-  MessageLoop message_loop_;
   scoped_ptr<ChromotingHostContext> context_;
   scoped_ptr<IPC::ChannelProxy> daemon_channel_;
   scoped_ptr<net::NetworkChangeNotifier> network_change_notifier_;
@@ -716,12 +737,11 @@ int main(int argc, char** argv) {
               logging::APPEND_TO_OLD_LOG_FILE,
               logging::DISABLE_DCHECK_FOR_NON_OFFICIAL_RELEASE_BUILDS);
 
-  const CommandLine* cmd_line = CommandLine::ForCurrentProcess();
-
 #if defined(TOOLKIT_GTK)
   // Required for any calls into GTK functions, such as the Disconnect and
   // Continue windows, though these should not be used for the Me2Me case
   // (crbug.com/104377).
+  const CommandLine* cmd_line = CommandLine::ForCurrentProcess();
   gfx::GtkInitFromCommandLine(*cmd_line);
 #endif  // TOOLKIT_GTK
 
@@ -730,15 +750,28 @@ int main(int argc, char** argv) {
   net::EnableSSLServerSockets();
 
 #if defined(OS_LINUX)
-    remoting::VideoFrameCapturer::EnableXDamage(true);
+  remoting::VideoFrameCapturer::EnableXDamage(true);
 #endif
 
-  remoting::HostProcess me2me_host;
-  if (!me2me_host.InitWithCommandLine(cmd_line)) {
-    return remoting::kInvalidHostConfigurationExitCode;
-  }
-
-  return me2me_host.Run();
+  // Create the main message loop and start helper threads.
+  MessageLoop message_loop(MessageLoop::TYPE_UI);
+  base::Closure quit_message_loop = base::Bind(&QuitMessageLoop, &message_loop);
+  scoped_ptr<remoting::ChromotingHostContext> context(
+      new remoting::ChromotingHostContext(
+          new remoting::AutoThreadTaskRunner(message_loop.message_loop_proxy(),
+                                             quit_message_loop)));
+  if (!context->Start())
+    return remoting::kHostInitializationFailed;
+
+  // Create the host process instance and run the rest of the initialization on
+  // the main message loop.
+  remoting::HostProcess me2me_host(context.Pass());
+  message_loop.PostTask(
+      FROM_HERE,
+      base::Bind(&remoting::HostProcess::StartHostProcess,
+                 base::Unretained(&me2me_host)));
+  message_loop.Run();
+  return me2me_host.get_exit_code();
 }
 
 #if defined(OS_WIN)
diff --git a/remoting/host/simple_host_process.cc b/remoting/host/simple_host_process.cc
index 6353a3d1..0887eb3 100644
--- a/remoting/host/simple_host_process.cc
+++ b/remoting/host/simple_host_process.cc
@@ -32,6 +32,7 @@
 #include "crypto/nss_util.h"
 #include "net/base/network_change_notifier.h"
 #include "net/socket/ssl_server_socket.h"
+#include "remoting/base/auto_thread_task_runner.h"
 #include "remoting/base/constants.h"
 #include "remoting/host/audio_capturer.h"
 #include "remoting/host/chromoting_host_context.h"
@@ -97,7 +98,7 @@ class SimpleHost : public HeartbeatSender::Listener {
  public:
   SimpleHost()
       : message_loop_(MessageLoop::TYPE_UI),
-        context_(message_loop_.message_loop_proxy()),
+        context_(new AutoThreadTaskRunner(message_loop_.message_loop_proxy())),
         fake_(false),
         is_it2me_(false),
         shutting_down_(false),
diff --git a/remoting/host/win/host_service.cc b/remoting/host/win/host_service.cc
index 87088fe..92dda1d 100644
--- a/remoting/host/win/host_service.cc
+++ b/remoting/host/win/host_service.cc
@@ -24,6 +24,7 @@
 #include "base/threading/thread.h"
 #include "base/utf_string_conversions.h"
 #include "base/win/wrapped_window_proc.h"
+#include "remoting/base/auto_thread_task_runner.h"
 #include "remoting/base/breakpad.h"
 #include "remoting/base/scoped_sc_handle_win.h"
 #include "remoting/base/stoppable.h"
@@ -92,6 +93,10 @@ void usage(const FilePath& program_name) {
                             UTF16ToWide(program_name.value()).c_str());
 }
 
+void QuitMessageLoop(MessageLoop* message_loop) {
+  message_loop->PostTask(FROM_HERE, MessageLoop::QuitClosure());
+}
+
 }  // namespace
 
 namespace remoting {
@@ -120,7 +125,7 @@ void HostService::RemoveWtsConsoleObserver(WtsConsoleObserver* observer) {
 
 void HostService::OnChildStopped() {
   child_.reset(NULL);
-  main_task_runner_->PostTask(FROM_HERE, MessageLoop::QuitClosure());
+  main_task_runner_ = NULL;
 }
 
 void HostService::OnSessionChange() {
@@ -196,21 +201,14 @@ int HostService::Run() {
   return (this->*run_routine_)();
 }
 
-void HostService::RunMessageLoop(MessageLoop* message_loop) {
-  // Launch the I/O thread.
-  base::Thread io_thread(kIoThreadName);
-  base::Thread::Options io_thread_options(MessageLoop::TYPE_IO, 0);
-  if (!io_thread.StartWithOptions(io_thread_options)) {
-    LOG(ERROR) << "Failed to start the I/O thread";
-    stopped_event_.Signal();
-    return;
-  }
+void HostService::CreateLauncher(
+    scoped_refptr<base::SingleThreadTaskRunner> io_task_runner) {
 
 #if defined(REMOTING_MULTI_PROCESS)
 
   child_ = DaemonProcess::Create(
       main_task_runner_,
-      io_thread.message_loop_proxy(),
+      io_task_runner,
       base::Bind(&HostService::OnChildStopped,
                  base::Unretained(this))).PassAs<Stoppable>();
 
@@ -221,15 +219,25 @@ void HostService::RunMessageLoop(MessageLoop* message_loop) {
       base::Bind(&HostService::OnChildStopped, base::Unretained(this)),
       this,
       main_task_runner_,
-      io_thread.message_loop_proxy()));
+      io_task_runner));
 
 #endif  // !defined(REMOTING_MULTI_PROCESS)
+}
+
+void HostService::RunMessageLoop(MessageLoop* message_loop) {
+  // Launch the I/O thread.
+  base::Thread io_thread(kIoThreadName);
+  base::Thread::Options io_thread_options(MessageLoop::TYPE_IO, 0);
+  if (!io_thread.StartWithOptions(io_thread_options)) {
+    LOG(FATAL) << "Failed to start the I/O thread";
+    return;
+  }
+
+  CreateLauncher(new AutoThreadTaskRunner(io_thread.message_loop_proxy(),
+                                          main_task_runner_));
 
   // Run the service.
   message_loop->Run();
-
-  // Release the control handler.
-  stopped_event_.Signal();
 }
 
 int HostService::Elevate() {
@@ -279,8 +287,11 @@ int HostService::RunAsService() {
 int HostService::RunInConsole() {
   MessageLoop message_loop(MessageLoop::TYPE_UI);
 
-  // Allow other threads to post to our message loop.
-  main_task_runner_ = message_loop.message_loop_proxy();
+  // Keep a reference to the main message loop while it is used. Once the last
+  // reference is dropped, QuitClosure() will be posted to the loop.
+  main_task_runner_ =
+      new AutoThreadTaskRunner(message_loop.message_loop_proxy(),
+                               base::Bind(&QuitMessageLoop, &message_loop));
 
   int result = kErrorExitCode;
 
@@ -327,6 +338,9 @@ int HostService::RunInConsole() {
     // Run the service.
     RunMessageLoop(&message_loop);
 
+    // Release the control handler.
+    stopped_event_.Signal();
+
     WTSUnRegisterSessionNotification(window);
     result = kSuccessExitCode;
   }
@@ -375,11 +389,14 @@ DWORD WINAPI HostService::ServiceControlHandler(DWORD control,
 }
 
 VOID WINAPI HostService::ServiceMain(DWORD argc, WCHAR* argv[]) {
-  MessageLoop message_loop;
+  MessageLoop message_loop(MessageLoop::TYPE_DEFAULT);
 
-  // Allow other threads to post to our message loop.
+  // Keep a reference to the main message loop while it is used. Once the last
+  // reference is dropped QuitClosure() will be posted to the loop.
   HostService* self = HostService::GetInstance();
-  self->main_task_runner_ = message_loop.message_loop_proxy();
+  self->main_task_runner_ =
+      new AutoThreadTaskRunner(message_loop.message_loop_proxy(),
+                               base::Bind(&QuitMessageLoop, &message_loop));
 
   // Register the service control handler.
   self->service_status_handle_ =
@@ -416,6 +433,9 @@ VOID WINAPI HostService::ServiceMain(DWORD argc, WCHAR* argv[]) {
   // Run the service.
   self->RunMessageLoop(&message_loop);
 
+  // Release the control handler.
+  self->stopped_event_.Signal();
+
   // Tell SCM that the service is stopped.
   service_status.dwCurrentState = SERVICE_STOPPED;
   service_status.dwControlsAccepted = 0;
diff --git a/remoting/host/win/host_service.h b/remoting/host/win/host_service.h
index 9533f02..80bbc90 100644
--- a/remoting/host/win/host_service.h
+++ b/remoting/host/win/host_service.h
@@ -22,14 +22,13 @@ class SingleThreadTaskRunner;
 
 namespace remoting {
 
-#if defined(REMOTING_MULTI_PROCESS)
-class DaemonProcess;
-#endif  // defined(REMOTING_MULTI_PROCESS)
-
+class AutoThreadTaskRunner;
 class Stoppable;
 class WtsConsoleObserver;
 
-#if !defined(REMOTING_MULTI_PROCESS)
+#if defined(REMOTING_MULTI_PROCESS)
+class DaemonProcess;
+#else  // !defined(REMOTING_MULTI_PROCESS)
 class WtsSessionProcessLauncher;
 #endif  // !defined(REMOTING_MULTI_PROCESS)
 
@@ -57,6 +56,10 @@ class HostService : public WtsConsoleMonitor {
   // Notifies the service of changes in session state.
   void OnSessionChange();
 
+  // Creates the process launcher.
+  void CreateLauncher(
+    scoped_refptr<base::SingleThreadTaskRunner> io_task_runner);
+
   // This is a common entry point to the main service loop called by both
   // RunAsService() and RunInConsole().
   void RunMessageLoop(MessageLoop* message_loop);
@@ -98,7 +101,7 @@ class HostService : public WtsConsoleMonitor {
   scoped_ptr<Stoppable> child_;
 
   // Service message loop.
-  scoped_refptr<base::SingleThreadTaskRunner> main_task_runner_;
+  scoped_refptr<AutoThreadTaskRunner> main_task_runner_;
 
   // The action routine to be executed.
   int (HostService::*run_routine_)();