Move content's shell connections to the IO thread

content/common/mojo/mojo_shell_connection_impl.cc

 // 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.
 
 #include "content/common/mojo/mojo_shell_connection_impl.h"
 
+#include <utility>
+#include <vector>
+
 #include "base/bind.h"
+#include "base/callback_helpers.h"
 #include "base/lazy_instance.h"
+#include "base/macros.h"
 #include "base/memory/ptr_util.h"
-#include "base/threading/thread_local.h"
+#include "base/threading/thread_checker.h"
 #include "content/common/mojo/embedded_application_runner.h"
+#include "mojo/public/cpp/bindings/binding_set.h"
+#include "mojo/public/cpp/system/message_pipe.h"
 #include "services/shell/public/cpp/shell_client.h"
 #include "services/shell/public/cpp/shell_connection.h"
+#include "services/shell/public/interfaces/shell_client_factory.mojom.h"
 #include "services/shell/runner/common/client_util.h"
 
 namespace content {
 namespace {
 
-using MojoShellConnectionPtr =
-    base::ThreadLocalPointer<MojoShellConnection>;
-
-// Env is thread local so that aura may be used on multiple threads.
-base::LazyInstance<MojoShellConnectionPtr>::Leaky lazy_tls_ptr =
-    LAZY_INSTANCE_INITIALIZER;
+base::LazyInstance<std::unique_ptr<MojoShellConnection>>::Leaky
+    g_connection_for_process = LAZY_INSTANCE_INITIALIZER;
 
 MojoShellConnection::Factory* mojo_shell_connection_factory = nullptr;
 
 }  // namespace
 
+// A ref-counted object which owns the IO thread state of a
+// MojoShellConnectionImpl. This includes ShellClient and ShellClientFactory
+// bindings.
+class MojoShellConnectionImpl::IOThreadContext
+    : public base::RefCountedThreadSafe<IOThreadContext>,
+      public shell::ShellClient,
+      public shell::InterfaceFactory<shell::mojom::ShellClientFactory>,
+      public shell::mojom::ShellClientFactory {
+ public:
+  using ShellClientFactoryCallback =
+      base::Callback<void(shell::mojom::ShellClientRequest,
+                          const mojo::String&)>;
+
+  IOThreadContext() {}
+
+  // The following methods are safe to call from any thread.
+
+  void Initialize(shell::mojom::ShellClientRequest request,
+                  scoped_refptr<base::SequencedTaskRunner> io_task_runner,
+                  std::unique_ptr<shell::Connector> io_thread_connector,
+                  shell::mojom::ConnectorRequest connector_request,
+                  const base::Closure& initialize_callback,
+                  const ShellClientFactoryCallback& create_shell_client) {
+    DCHECK(!io_task_runner_);
+    io_thread_checker_.DetachFromThread();
+
+    io_task_runner_ = io_task_runner;
+    initialize_handler_ = base::Bind(
+        &IOThreadContext::CallInitializeOnTaskRunner,
+        base::ThreadTaskRunnerHandle::Get(), initialize_callback);
+
+    create_shell_client_ = base::Bind(
+        &IOThreadContext::CallShellClientFactoryOnTaskRunner,
+        base::ThreadTaskRunnerHandle::Get(), create_shell_client);
+
+    io_task_runner_->PostTask(
+        FROM_HERE,
+        base::Bind(&IOThreadContext::InitializeOnIOThread, this,
+                   base::Passed(&request), base::Passed(&io_thread_connector),
+                   base::Passed(&connector_request)));
+  }
+
+  void ShutDown() {
+    bool posted = io_task_runner_->PostTask(
+        FROM_HERE, base::Bind(&IOThreadContext::ShutdownOnIOThread, this));
+
+    // This must be called before IO thread shutdown.
+    DCHECK(posted);
+  }
+
+  void AddConnectionFilter(
+      const MojoShellConnection::ConnectionFilter& filter) {
+    base::AutoLock lock(lock_);
+    connection_filters_.push_back(filter);
+  }
+
+  shell::Identity GetIdentity() {
+    base::AutoLock lock(lock_);
+    if (!shell_connection_)
+      return shell::Identity();
+
+    return shell_connection_->identity();
+  }
+
+ private:
+  friend class base::RefCountedThreadSafe<IOThreadContext>;
+
+  ~IOThreadContext() override {}
+
+  void InitializeOnIOThread(shell::mojom::ShellClientRequest request,
+                            std::unique_ptr<shell::Connector> connector,
+                            shell::mojom::ConnectorRequest connector_request) {
+    // Should bind |io_thread_checker_| to the context's thread.
+    DCHECK(io_thread_checker_.CalledOnValidThread());
+    shell_connection_.reset(new shell::ShellConnection(
+        this, std::move(request), std::move(connector),
+        std::move(connector_request)));
+  }
+
+  void ShutdownOnIOThread() {
+    DCHECK(io_thread_checker_.CalledOnValidThread());
+    shell_connection_.reset();
+    factory_bindings_.CloseAllBindings();
+  }
+
+  /////////////////////////////////////////////////////////////////////////////
+  // shell::ShellClient implementation
+
+  void Initialize(shell::Connector* connector,
+                  const shell::Identity& identity,
+                  uint32_t id) override {
+    DCHECK(io_thread_checker_.CalledOnValidThread());
+    DCHECK(!initialize_handler_.is_null());
+    base::ResetAndReturn(&initialize_handler_).Run();
+  }
+
+  bool AcceptConnection(shell::Connection* connection) override {
+    DCHECK(io_thread_checker_.CalledOnValidThread());
+    std::string remote_app = connection->GetRemoteIdentity().name();
+    if (remote_app == "mojo:shell") {
+      // Only expose the SCF interface to the shell.
+      connection->AddInterface<shell::mojom::ShellClientFactory>(this);
+      return true;
+    }
+
+    bool accept = false;
+    base::AutoLock lock(lock_);
+    for (auto& filter : connection_filters_)
+      accept = accept || filter.Run(connection, shell_connection_->connector());
+
+    // Reject all other connections to this application.
+    return accept;
+  }
+
+  /////////////////////////////////////////////////////////////////////////////
+  // shell::InterfaceFactory<shell::mojom::ShellClientFactory> implementation
+
+  void Create(shell::Connection* connection,
+              shell::mojom::ShellClientFactoryRequest request) override {
+    DCHECK(io_thread_checker_.CalledOnValidThread());
+    factory_bindings_.AddBinding(this, std::move(request));
+  }
+
+  /////////////////////////////////////////////////////////////////////////////
+  // shell::mojom::ShellClientFactory implementation
+
+  void CreateShellClient(shell::mojom::ShellClientRequest request,
+                         const mojo::String& name) override {
+    DCHECK(io_thread_checker_.CalledOnValidThread());
+    create_shell_client_.Run(std::move(request), name);
+  }
+
+  static void CallInitializeOnTaskRunner(
+      scoped_refptr<base::SequencedTaskRunner> task_runner,
+      const base::Closure& callback) {
+    task_runner->PostTask(FROM_HERE, callback);
+  }
+
+  static void CallShellClientFactoryOnTaskRunner(
+      scoped_refptr<base::SequencedTaskRunner> task_runner,
+      const ShellClientFactoryCallback& callback,
+      shell::mojom::ShellClientRequest request,
+      const mojo::String& name) {
+    task_runner->PostTask(
+        FROM_HERE, base::Bind(callback, base::Passed(&request), name));
+  }
+
+  base::ThreadChecker io_thread_checker_;
+
+  scoped_refptr<base::SequencedTaskRunner> io_task_runner_;
+
+  base::Closure initialize_handler_;
+  ShellClientFactoryCallback create_shell_client_;
+
+  // Guards fields below for thread-safe access.
+  base::Lock lock_;
+
+  std::unique_ptr<shell::ShellConnection> shell_connection_;
+  mojo::BindingSet<shell::mojom::ShellClientFactory> factory_bindings_;
+  std::vector<MojoShellConnection::ConnectionFilter> connection_filters_;
+
+  DISALLOW_COPY_AND_ASSIGN(IOThreadContext);
+};
+
 ////////////////////////////////////////////////////////////////////////////////
 // MojoShellConnection, public:
 
 // static
 void MojoShellConnection::SetForProcess(
     std::unique_ptr<MojoShellConnection> connection) {
-  DCHECK(!lazy_tls_ptr.Pointer()->Get());
-  lazy_tls_ptr.Pointer()->Set(connection.release());
+  DCHECK(!g_connection_for_process.Get());
+  g_connection_for_process.Get() = std::move(connection);
 }
 
 // static
 MojoShellConnection* MojoShellConnection::GetForProcess() {
-  return lazy_tls_ptr.Pointer()->Get();
+  return g_connection_for_process.Get().get();
 }
 
 // static
 void MojoShellConnection::DestroyForProcess() {
   // This joins the shell controller thread.
-  delete GetForProcess();
-  lazy_tls_ptr.Pointer()->Set(nullptr);
+  g_connection_for_process.Get().reset();
 }
 
 // static
 void MojoShellConnection::SetFactoryForTest(Factory* factory) {
-  DCHECK(!lazy_tls_ptr.Pointer()->Get());
+  DCHECK(!g_connection_for_process.Get());
   mojo_shell_connection_factory = factory;
 }
 
 // static
 std::unique_ptr<MojoShellConnection> MojoShellConnection::Create(
-    shell::mojom::ShellClientRequest request) {
+    shell::mojom::ShellClientRequest request,
+    scoped_refptr<base::SequencedTaskRunner> io_task_runner) {
   if (mojo_shell_connection_factory)
     return mojo_shell_connection_factory->Run();
-  return base::WrapUnique(new MojoShellConnectionImpl(std::move(request)));
+  return base::MakeUnique<MojoShellConnectionImpl>(
+      std::move(request), io_task_runner);
 }
 
 MojoShellConnection::~MojoShellConnection() {}
 
 ////////////////////////////////////////////////////////////////////////////////
 // MojoShellConnectionImpl, public:
 
 MojoShellConnectionImpl::MojoShellConnectionImpl(
-    shell::mojom::ShellClientRequest request)
-    : shell_connection_(new shell::ShellConnection(this, std::move(request))) {}
+    shell::mojom::ShellClientRequest request,
+    scoped_refptr<base::SequencedTaskRunner> io_task_runner)
+    : context_(new IOThreadContext), weak_factory_(this) {
+  shell::mojom::ConnectorRequest connector_request;
+  connector_ = shell::Connector::Create(&connector_request);
 
-MojoShellConnectionImpl::~MojoShellConnectionImpl() {}
-
-////////////////////////////////////////////////////////////////////////////////
-// MojoShellConnectionImpl, shell::ShellClient implementation:
-
-void MojoShellConnectionImpl::Initialize(shell::Connector* connector,
-                                         const shell::Identity& identity,
-                                         uint32_t id) {
-  for (auto& client : embedded_shell_clients_)
-    client->Initialize(connector, identity, id);
+  std::unique_ptr<shell::Connector> io_thread_connector = connector_->Clone();
+  context_->Initialize(
+      std::move(request), io_task_runner, std::move(io_thread_connector),
+      std::move(connector_request),
+      base::Bind(&MojoShellConnectionImpl::OnContextInitialized,
+                 weak_factory_.GetWeakPtr()),
+      base::Bind(&MojoShellConnectionImpl::CreateShellClient,
+                 weak_factory_.GetWeakPtr()));
 }
 
-bool MojoShellConnectionImpl::AcceptConnection(shell::Connection* connection) {
-  std::string remote_app = connection->GetRemoteIdentity().name();
-  if (remote_app == "mojo:shell") {
-    // Only expose the SCF interface to the shell.
-    connection->AddInterface<shell::mojom::ShellClientFactory>(this);
-    return true;
-  }
-
-  bool accept = false;
-  for (auto& client : embedded_shell_clients_)
-    accept |= client->AcceptConnection(connection);
-
-  // Reject all other connections to this application.
-  return accept;
-}
-
-shell::InterfaceRegistry*
-MojoShellConnectionImpl::GetInterfaceRegistryForConnection() {
-  // TODO(beng): This is really horrible since obviously subject to issues
-  // of ordering, but is no more horrible than this API is in general.
-  shell::InterfaceRegistry* registry = nullptr;
-  for (auto& client : embedded_shell_clients_) {
-    registry = client->GetInterfaceRegistryForConnection();
-    if (registry)
-      return registry;
-  }
-  return nullptr;
-}
-
-shell::InterfaceProvider*
-MojoShellConnectionImpl::GetInterfaceProviderForConnection() {
-  // TODO(beng): This is really horrible since obviously subject to issues
-  // of ordering, but is no more horrible than this API is in general.
-  shell::InterfaceProvider* provider = nullptr;
-  for (auto& client : embedded_shell_clients_) {
-    provider = client->GetInterfaceProviderForConnection();
-    if (provider)
-      return provider;
-  }
-  return nullptr;
-}
-
-////////////////////////////////////////////////////////////////////////////////
-// MojoShellConnectionImpl,
-//     shell::InterfaceFactory<shell::mojom::ShellClientFactory> implementation:
-
-void MojoShellConnectionImpl::Create(
-    shell::Connection* connection,
-    shell::mojom::ShellClientFactoryRequest request) {
-  factory_bindings_.AddBinding(this, std::move(request));
-}
-
-////////////////////////////////////////////////////////////////////////////////
-// MojoShellConnectionImpl, shell::mojom::ShellClientFactory implementation:
-
-void MojoShellConnectionImpl::CreateShellClient(
-    shell::mojom::ShellClientRequest request,
-    const mojo::String& name) {
-  auto it = request_handlers_.find(name);
-  if (it != request_handlers_.end())
-    it->second.Run(std::move(request));
+MojoShellConnectionImpl::~MojoShellConnectionImpl() {
+  context_->ShutDown();
 }
 
 ////////////////////////////////////////////////////////////////////////////////
 // MojoShellConnectionImpl, MojoShellConnection implementation:
 
-shell::ShellConnection* MojoShellConnectionImpl::GetShellConnection() {
-  return shell_connection_.get();
+void MojoShellConnectionImpl::SetInitializeHandler(
+    const base::Closure& handler) {
+  DCHECK(initialize_handler_.is_null());
+  initialize_handler_ = handler;
 }
 
 shell::Connector* MojoShellConnectionImpl::GetConnector() {
-  DCHECK(shell_connection_);
-  return shell_connection_->connector();
+  return connector_.get();
 }
 
 const shell::Identity& MojoShellConnectionImpl::GetIdentity() const {
-  DCHECK(shell_connection_);
-  return shell_connection_->identity();
+  return identity_;
 }
 
 void MojoShellConnectionImpl::SetConnectionLostClosure(
     const base::Closure& closure) {
-  shell_connection_->SetConnectionLostClosure(closure);
+  connection_lost_handler_ = closure;
 }
 
-void MojoShellConnectionImpl::AddEmbeddedShellClient(
-    std::unique_ptr<shell::ShellClient> shell_client) {
-  embedded_shell_clients_.push_back(shell_client.get());
-  owned_shell_clients_.push_back(std::move(shell_client));
-}
-
-void MojoShellConnectionImpl::AddEmbeddedShellClient(
-    shell::ShellClient* shell_client) {
-  embedded_shell_clients_.push_back(shell_client);
+void MojoShellConnectionImpl::AddConnectionFilter(
+    const ConnectionFilter& filter) {

[Ben Goodger (Google), 2016/07/03 03:29:27]

did you mean to leave this implementation empty?

 }
 
 void MojoShellConnectionImpl::AddEmbeddedService(
     const std::string& name,
     const MojoApplicationInfo& info) {
   std::unique_ptr<EmbeddedApplicationRunner> app(
       new EmbeddedApplicationRunner(name, info));
   AddShellClientRequestHandler(
       name, base::Bind(&EmbeddedApplicationRunner::BindShellClientRequest,
                        base::Unretained(app.get())));
   auto result = embedded_apps_.insert(std::make_pair(name, std::move(app)));
   DCHECK(result.second);
 }
 
 void MojoShellConnectionImpl::AddShellClientRequestHandler(
     const std::string& name,
     const ShellClientRequestHandler& handler) {
   auto result = request_handlers_.insert(std::make_pair(name, handler));
   DCHECK(result.second);
 }
 
+void MojoShellConnectionImpl::CreateShellClient(
+    shell::mojom::ShellClientRequest request,
+    const mojo::String& name) {
+  auto it = request_handlers_.find(name);
+  if (it != request_handlers_.end())
+    it->second.Run(std::move(request));
+}
+
+void MojoShellConnectionImpl::OnContextInitialized() {
+  identity_ = context_->GetIdentity();
+  if (!initialize_handler_.is_null())
+    base::ResetAndReturn(&initialize_handler_).Run();
+}
+
+void MojoShellConnectionImpl::OnConnectionLost() {
+  if (!connection_lost_handler_.is_null())
+    connection_lost_handler_.Run();
+}
+
 }  // namespace content