Fixing threading issues in remoting::DesktopProcess. The UI thread is now owned by the caller of remoting::DesktopThread so it can be properly shared between the caller and the class.

BUG=157808,157886

Review URL: https://codereview.chromium.org/11272036

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

4 files changed, 132 insertions, 82 deletions

diff --git a/remoting/host/desktop_process.cc b/remoting/host/desktop_process.cc
index d2d25ac..7f3d730 100644
--- a/remoting/host/desktop_process.cc
+++ b/remoting/host/desktop_process.cc
@@ -12,26 +12,33 @@
 #include "base/logging.h"
 #include "base/memory/ref_counted.h"
 #include "base/message_loop.h"
-#include "base/run_loop.h"
 #include "ipc/ipc_channel_proxy.h"
 #include "remoting/base/auto_thread.h"
 #include "remoting/base/auto_thread_task_runner.h"
 #include "remoting/host/chromoting_messages.h"
 #include "remoting/host/desktop_session_agent.h"
-#include "remoting/host/host_exit_codes.h"
 
 const char kIoThreadName[] = "I/O thread";
 
 namespace remoting {
 
-DesktopProcess::DesktopProcess(const std::string& daemon_channel_name)
-    : daemon_channel_name_(daemon_channel_name) {
+DesktopProcess::DesktopProcess(
+    scoped_refptr<AutoThreadTaskRunner> caller_task_runner,
+    const std::string& daemon_channel_name)
+    : caller_task_runner_(caller_task_runner),
+      daemon_channel_name_(daemon_channel_name) {
+  DCHECK(caller_task_runner_->BelongsToCurrentThread());
+  DCHECK_EQ(MessageLoop::current()->type(), MessageLoop::TYPE_UI);
 }
 
 DesktopProcess::~DesktopProcess() {
+  DCHECK(!daemon_channel_);
+  DCHECK(!desktop_agent_);
 }
 
 bool DesktopProcess::OnMessageReceived(const IPC::Message& message) {
+  DCHECK(caller_task_runner_->BelongsToCurrentThread());
+
   bool handled = true;
   IPC_BEGIN_MESSAGE_MAP(DesktopProcess, message)
     IPC_MESSAGE_HANDLER(ChromotingDaemonDesktopMsg_Crash, OnCrash)
@@ -41,64 +48,56 @@ bool DesktopProcess::OnMessageReceived(const IPC::Message& message) {
 }
 
 void DesktopProcess::OnChannelConnected(int32 peer_pid) {
+  DCHECK(caller_task_runner_->BelongsToCurrentThread());
+
   VLOG(1) << "IPC: desktop <- daemon (" << peer_pid << ")";
 
   NOTIMPLEMENTED();
 }
 
 void DesktopProcess::OnChannelError() {
+  DCHECK(caller_task_runner_->BelongsToCurrentThread());
+
   // Shutdown the desktop process.
   daemon_channel_.reset();
   desktop_agent_.reset();
+  caller_task_runner_ = NULL;
 }
 
-int DesktopProcess::Run() {
-  DCHECK_EQ(MessageLoop::current()->type(), MessageLoop::TYPE_UI);
-
-  {
-    scoped_refptr<AutoThreadTaskRunner> ui_task_runner =
-        new remoting::AutoThreadTaskRunner(
-            MessageLoop::current()->message_loop_proxy(),
-            MessageLoop::QuitClosure());
-
-    // Launch the I/O thread.
-    scoped_refptr<AutoThreadTaskRunner> io_task_runner =
-        AutoThread::CreateWithType(kIoThreadName, ui_task_runner,
-                                   MessageLoop::TYPE_IO);
-
-    // Create a desktop agent.
-    desktop_agent_ = DesktopSessionAgent::Create(ui_task_runner,
-                                                 io_task_runner);
-
-    // Start the agent and create an IPC channel to talk to it. It is safe to
-    // use base::Unretained(this) here because the message loop below will run
-    // until |desktop_agent_| is completely destroyed.
-    IPC::PlatformFileForTransit desktop_pipe;
-    if (!desktop_agent_->Start(base::Bind(&DesktopProcess::OnChannelError,
-                                          base::Unretained(this)),
-                               &desktop_pipe)) {
-      desktop_agent_.reset();
-      return kInitializationFailed;
-    }
-
-    // Connect to the daemon.
-    daemon_channel_.reset(new IPC::ChannelProxy(daemon_channel_name_,
-                                                IPC::Channel::MODE_CLIENT,
-                                                this,
-                                                io_task_runner));
-
-    // Pass |desktop_pipe| to the daemon.
-    daemon_channel_->Send(
-        new ChromotingDesktopDaemonMsg_DesktopAttached(desktop_pipe));
+bool DesktopProcess::Start() {
+  DCHECK(caller_task_runner_->BelongsToCurrentThread());
+
+  // Launch the I/O thread.
+  scoped_refptr<AutoThreadTaskRunner> io_task_runner =
+      AutoThread::CreateWithType(kIoThreadName, caller_task_runner_,
+                                 MessageLoop::TYPE_IO);
+
+  // Create a desktop agent.
+  desktop_agent_ = DesktopSessionAgent::Create(caller_task_runner_,
+                                               io_task_runner);
+
+  // Start the agent and create an IPC channel to talk to it. It is safe to
+  // use base::Unretained(this) here because the message loop below will run
+  // until |desktop_agent_| is completely destroyed.
+  IPC::PlatformFileForTransit desktop_pipe;
+  if (!desktop_agent_->Start(base::Bind(&DesktopProcess::OnChannelError,
+                                        base::Unretained(this)),
+                             &desktop_pipe)) {
+    desktop_agent_.reset();
+    return false;
   }
 
-  // Run the UI message loop.
-  base::RunLoop run_loop;
-  run_loop.Run();
+  // Connect to the daemon.
+  daemon_channel_.reset(new IPC::ChannelProxy(daemon_channel_name_,
+                                              IPC::Channel::MODE_CLIENT,
+                                              this,
+                                              io_task_runner));
 
-  DCHECK(!daemon_channel_);
-  DCHECK(!desktop_agent_);
-  return kSuccessExitCode;
+  // Pass |desktop_pipe| to the daemon.
+  daemon_channel_->Send(
+      new ChromotingDesktopDaemonMsg_DesktopAttached(desktop_pipe));
+
+  return true;
 }
 
 void DesktopProcess::OnCrash(const std::string& function_name,
diff --git a/remoting/host/desktop_process.h b/remoting/host/desktop_process.h
index 9226407..75786b5 100644
--- a/remoting/host/desktop_process.h
+++ b/remoting/host/desktop_process.h
@@ -9,6 +9,7 @@
 
 #include "base/basictypes.h"
 #include "base/compiler_specific.h"
+#include "base/memory/ref_counted.h"
 #include "base/memory/scoped_ptr.h"
 #include "ipc/ipc_listener.h"
 
@@ -18,11 +19,13 @@ class ChannelProxy;
 
 namespace remoting {
 
+class AutoThreadTaskRunner;
 class DesktopSessionAgent;
 
 class DesktopProcess : public IPC::Listener {
  public:
-  explicit DesktopProcess(const std::string& daemon_channel_name);
+  DesktopProcess(scoped_refptr<AutoThreadTaskRunner> caller_task_runner,
+                 const std::string& daemon_channel_name);
   virtual ~DesktopProcess();
 
   // IPC::Listener implementation.
@@ -30,8 +33,9 @@ class DesktopProcess : public IPC::Listener {
   virtual void OnChannelConnected(int32 peer_pid) OVERRIDE;
   virtual void OnChannelError() OVERRIDE;
 
-  // Runs the desktop process.
-  int Run();
+  // Creates the desktop agent and required threads and IPC channels. Returns
+  // true on success.
+  bool Start();
 
  private:
   // Crashes the process in response to a daemon's request. The daemon passes
@@ -41,6 +45,9 @@ class DesktopProcess : public IPC::Listener {
                const std::string& file_name,
                const int& line_number);
 
+  // Task runner on which public methods of this class should be called.
+  scoped_refptr<AutoThreadTaskRunner> caller_task_runner_;
+
   // Name of the IPC channel connecting the desktop process with the daemon
   // process.
   std::string daemon_channel_name_;
diff --git a/remoting/host/desktop_process_main.cc b/remoting/host/desktop_process_main.cc
index bdfb8af..c7b6a0d 100644
--- a/remoting/host/desktop_process_main.cc
+++ b/remoting/host/desktop_process_main.cc
@@ -6,12 +6,17 @@
 // running within user sessions.
 
 #include "base/at_exit.h"
+#include "base/bind.h"
+#include "base/bind_helpers.h"
 #include "base/command_line.h"
 #include "base/file_path.h"
+#include "base/message_loop.h"
+#include "base/run_loop.h"
 #include "base/scoped_native_library.h"
 #include "base/stringprintf.h"
 #include "base/utf_string_conversions.h"
 #include "base/win/windows_version.h"
+#include "remoting/base/auto_thread_task_runner.h"
 #include "remoting/host/desktop_process.h"
 #include "remoting/host/host_exit_codes.h"
 #include "remoting/host/logging.h"
@@ -77,8 +82,25 @@ int main(int argc, char** argv) {
     return remoting::kUsageExitCode;
   }
 
-  remoting::DesktopProcess desktop_process(channel_name);
-  return desktop_process.Run();
+  MessageLoop message_loop(MessageLoop::TYPE_UI);
+  base::RunLoop run_loop;
+  base::Closure quit_ui_task_runner = base::Bind(
+      base::IgnoreResult(&base::SingleThreadTaskRunner::PostTask),
+      message_loop.message_loop_proxy(),
+      FROM_HERE, run_loop.QuitClosure());
+  scoped_refptr<remoting::AutoThreadTaskRunner> ui_task_runner =
+      new remoting::AutoThreadTaskRunner(message_loop.message_loop_proxy(),
+                                         quit_ui_task_runner);
+
+  remoting::DesktopProcess desktop_process(ui_task_runner, channel_name);
+  if (!desktop_process.Start())
+    return remoting::kInitializationFailed;
+
+  // Run the UI message loop.
+  ui_task_runner = NULL;
+  run_loop.Run();
+
+  return remoting::kSuccessExitCode;
 }
 
 #if defined(OS_WIN)
diff --git a/remoting/host/desktop_process_unittest.cc b/remoting/host/desktop_process_unittest.cc
index 3849d0a..f9a8f13 100644
--- a/remoting/host/desktop_process_unittest.cc
+++ b/remoting/host/desktop_process_unittest.cc
@@ -4,9 +4,12 @@
 
 #include "remoting/host/desktop_process.h"
 
+#include "base/bind.h"
+#include "base/bind_helpers.h"
 #include "base/location.h"
 #include "base/memory/ref_counted.h"
 #include "base/message_loop.h"
+#include "base/run_loop.h"
 #include "base/single_thread_task_runner.h"
 #include "ipc/ipc_channel.h"
 #include "ipc/ipc_channel_proxy.h"
@@ -44,6 +47,21 @@ class MockDaemonListener : public IPC::Listener {
   DISALLOW_COPY_AND_ASSIGN(MockDaemonListener);
 };
 
+class MockNetworkListener : public IPC::Listener {
+ public:
+  MockNetworkListener() {}
+  virtual ~MockNetworkListener() {}
+
+  virtual bool OnMessageReceived(const IPC::Message& message) OVERRIDE;
+
+  MOCK_METHOD1(OnDesktopAttached, void(IPC::PlatformFileForTransit));
+  MOCK_METHOD1(OnChannelConnected, void(int32));
+  MOCK_METHOD0(OnChannelError, void());
+
+ private:
+  DISALLOW_COPY_AND_ASSIGN(MockNetworkListener);
+};
+
 bool MockDaemonListener::OnMessageReceived(const IPC::Message& message) {
   bool handled = true;
   IPC_BEGIN_MESSAGE_MAP(MockDaemonListener, message)
@@ -56,6 +74,15 @@ bool MockDaemonListener::OnMessageReceived(const IPC::Message& message) {
   return handled;
 }
 
+bool MockNetworkListener::OnMessageReceived(const IPC::Message& message) {
+  bool handled = true;
+
+  // TODO(alexeypa): handle received messages here.
+
+  EXPECT_TRUE(handled);
+  return handled;
+}
+
 }  // namespace
 
 class DesktopProcessTest : public testing::Test {
@@ -75,8 +102,6 @@ class DesktopProcessTest : public testing::Test {
   // exit.
   void DisconnectChannels();
 
-  void QuitMessageLoop();
-
   // Runs the desktop process code in a separate thread.
   void RunDesktopProcess();
 
@@ -87,9 +112,6 @@ class DesktopProcessTest : public testing::Test {
   void SendCrashRequest();
 
  protected:
-  // Name of the daemon-to_desktop channel.
-  std::string channel_name_;
-
   // The daemon's end of the daemon-to-desktop channel.
   scoped_ptr<IPC::ChannelProxy> daemon_channel_;
 
@@ -105,7 +127,7 @@ class DesktopProcessTest : public testing::Test {
   scoped_ptr<IPC::ChannelProxy> network_channel_;
 
   // Delegate that is passed to |network_channel_|.
-  MockDaemonListener network_listener_;
+  MockNetworkListener network_listener_;
 };
 
 
@@ -117,21 +139,6 @@ DesktopProcessTest::~DesktopProcessTest() {
 }
 
 void DesktopProcessTest::SetUp() {
-  scoped_refptr<AutoThreadTaskRunner> ui_task_runner = new AutoThreadTaskRunner(
-      message_loop_.message_loop_proxy(),
-      base::Bind(&DesktopProcessTest::QuitMessageLoop,
-                 base::Unretained(this)));
-
-  io_task_runner_ = AutoThread::CreateWithType("IPC thread", ui_task_runner,
-                                               MessageLoop::TYPE_IO);
-
-  // Create the daemon-to-desktop process channel.
-  channel_name_ = IPC::Channel::GenerateUniqueRandomChannelID();
-  daemon_channel_.reset(new IPC::ChannelProxy(
-      IPC::ChannelHandle(channel_name_),
-      IPC::Channel::MODE_SERVER,
-      &daemon_listener_,
-      io_task_runner_));
 }
 
 void DesktopProcessTest::TearDown() {
@@ -168,13 +175,30 @@ void DesktopProcessTest::DisconnectChannels() {
   io_task_runner_ = NULL;
 }
 
-void DesktopProcessTest::QuitMessageLoop() {
-  message_loop_.PostTask(FROM_HERE, MessageLoop::QuitClosure());
-}
-
 void DesktopProcessTest::RunDesktopProcess() {
-  DesktopProcess desktop_process(channel_name_);
-  EXPECT_EQ(desktop_process.Run(), kSuccessExitCode);
+  base::RunLoop run_loop;
+  base::Closure quit_ui_task_runner = base::Bind(
+      base::IgnoreResult(&base::SingleThreadTaskRunner::PostTask),
+      message_loop_.message_loop_proxy(),
+      FROM_HERE, run_loop.QuitClosure());
+  scoped_refptr<AutoThreadTaskRunner> ui_task_runner = new AutoThreadTaskRunner(
+      message_loop_.message_loop_proxy(), quit_ui_task_runner);
+
+  io_task_runner_ = AutoThread::CreateWithType("IPC thread", ui_task_runner,
+                                               MessageLoop::TYPE_IO);
+
+  std::string channel_name = IPC::Channel::GenerateUniqueRandomChannelID();
+  daemon_channel_.reset(new IPC::ChannelProxy(
+      IPC::ChannelHandle(channel_name),
+      IPC::Channel::MODE_SERVER,
+      &daemon_listener_,
+      io_task_runner_));
+
+  DesktopProcess desktop_process(ui_task_runner, channel_name);
+  EXPECT_TRUE(desktop_process.Start());
+
+  ui_task_runner = NULL;
+  run_loop.Run();
 }
 
 void DesktopProcessTest::RunDeathTest() {
@@ -224,8 +248,6 @@ TEST_F(DesktopProcessTest, DeathTest) {
   testing::GTEST_FLAG(death_test_style) = "threadsafe";
 
   EXPECT_DEATH(RunDeathTest(), "");
-
-  DisconnectChannels();
 }
 
 }  // namespace remoting