Mojo IPC threading fixes

ipc/mojo/ipc_channel_mojo.cc

 // Copyright 2014 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 "ipc/mojo/ipc_channel_mojo.h"
 
 #include "base/bind.h"
 #include "base/bind_helpers.h"
 #include "base/lazy_instance.h"
+#include "base/thread_task_runner_handle.h"
 #include "ipc/ipc_listener.h"
 #include "ipc/ipc_logging.h"
 #include "ipc/ipc_message_attachment_set.h"
 #include "ipc/ipc_message_macros.h"
 #include "ipc/mojo/client_channel.mojom.h"
 #include "ipc/mojo/ipc_mojo_bootstrap.h"
 #include "ipc/mojo/ipc_mojo_handle_attachment.h"
 #include "third_party/mojo/src/mojo/edk/embedder/embedder.h"
 #include "third_party/mojo/src/mojo/public/cpp/bindings/error_handler.h"
 
@@ skipping 47 matching lines @@
   void OnPipeAvailable(mojo::embedder::ScopedPlatformHandle handle) override;
   // mojo::ErrorHandler implementation
   void OnConnectionError() override;
   // ClientChannel implementation
   void Init(
       mojo::ScopedMessagePipeHandle pipe,
       int32_t peer_pid,
       const mojo::Callback<void(int32_t)>& callback) override;
 
  private:
+  void BindPipe(mojo::ScopedMessagePipeHandle handle);
+
   mojo::Binding<ClientChannel> binding_;
+  base::WeakPtrFactory<ClientChannelMojo> weak_factory_;
 
   DISALLOW_COPY_AND_ASSIGN(ClientChannelMojo);
 };
 
 ClientChannelMojo::ClientChannelMojo(ChannelMojo::Delegate* delegate,
                                      scoped_refptr<base::TaskRunner> io_runner,
                                      const ChannelHandle& handle,
                                      Listener* listener)
     : ChannelMojo(delegate, io_runner, handle, Channel::MODE_CLIENT, listener),
-      binding_(this) {
+      binding_(this),
+      weak_factory_(this) {
 }
 
 ClientChannelMojo::~ClientChannelMojo() {
 }
 
 void ClientChannelMojo::OnPipeAvailable(
     mojo::embedder::ScopedPlatformHandle handle) {
-  binding_.Bind(CreateMessagingPipe(handle.Pass()));
+  CreateMessagingPipe(handle.Pass(), base::Bind(&ClientChannelMojo::BindPipe,
+                                                weak_factory_.GetWeakPtr()));
 }
 
 void ClientChannelMojo::OnConnectionError() {
   listener()->OnChannelError();
 }
 
 void ClientChannelMojo::Init(
     mojo::ScopedMessagePipeHandle pipe,
     int32_t peer_pid,
     const mojo::Callback<void(int32_t)>& callback) {
   InitMessageReader(pipe.Pass(), static_cast<base::ProcessId>(peer_pid));
   callback.Run(GetSelfPID());
 }
 
+void ClientChannelMojo::BindPipe(mojo::ScopedMessagePipeHandle handle) {
+  binding_.Bind(handle.Pass());
+}
+
 //------------------------------------------------------------------------------
 
 class ServerChannelMojo : public ChannelMojo, public mojo::ErrorHandler {
  public:
   ServerChannelMojo(ChannelMojo::Delegate* delegate,
                     scoped_refptr<base::TaskRunner> io_runner,
                     const ChannelHandle& handle,
                     Listener* listener);
   ~ServerChannelMojo() override;
 
   // MojoBootstrap::Delegate implementation
   void OnPipeAvailable(mojo::embedder::ScopedPlatformHandle handle) override;
   // mojo::ErrorHandler implementation
   void OnConnectionError() override;
   // Channel override
   void Close() override;
 
  private:
+  void InitClientChannel(mojo::ScopedMessagePipeHandle peer_handle,
+                         mojo::ScopedMessagePipeHandle handle);
+
   // ClientChannelClient implementation
   void ClientChannelWasInitialized(int32_t peer_pid);
 
   mojo::InterfacePtr<ClientChannel> client_channel_;
   mojo::ScopedMessagePipeHandle message_pipe_;
+  base::WeakPtrFactory<ServerChannelMojo> weak_factory_;

[Hajime Morrita, 2015/05/11 22:19:21]

Having weak_factory_ for Channel is tricky but I believe this is safe
as we guarantee that the Channel is delete on IO thread...

 
   DISALLOW_COPY_AND_ASSIGN(ServerChannelMojo);
 };
 
 ServerChannelMojo::ServerChannelMojo(ChannelMojo::Delegate* delegate,
                                      scoped_refptr<base::TaskRunner> io_runner,
                                      const ChannelHandle& handle,
                                      Listener* listener)
-    : ChannelMojo(delegate, io_runner, handle, Channel::MODE_SERVER, listener) {
+    : ChannelMojo(delegate, io_runner, handle, Channel::MODE_SERVER, listener),
+      weak_factory_(this) {
 }
 
 ServerChannelMojo::~ServerChannelMojo() {
   Close();
 }
 
 void ServerChannelMojo::OnPipeAvailable(
     mojo::embedder::ScopedPlatformHandle handle) {
   mojo::ScopedMessagePipeHandle peer;
   MojoResult create_result =
       mojo::CreateMessagePipe(nullptr, &message_pipe_, &peer);
   if (create_result != MOJO_RESULT_OK) {
     LOG(WARNING) << "mojo::CreateMessagePipe failed: " << create_result;
     listener()->OnChannelError();
     return;
   }
+  CreateMessagingPipe(
+      handle.Pass(),
+      base::Bind(&ServerChannelMojo::InitClientChannel,
+                 weak_factory_.GetWeakPtr(), base::Passed(&peer)));
+}
 
+void ServerChannelMojo::InitClientChannel(
+    mojo::ScopedMessagePipeHandle peer_handle,
+    mojo::ScopedMessagePipeHandle handle) {
   client_channel_.Bind(
-      mojo::InterfacePtrInfo<ClientChannel>(
-          CreateMessagingPipe(handle.Pass()), 0u));
+      mojo::InterfacePtrInfo<ClientChannel>(handle.Pass(), 0u));
   client_channel_.set_error_handler(this);
   client_channel_->Init(
-      peer.Pass(),
-      static_cast<int32_t>(GetSelfPID()),
+      peer_handle.Pass(), static_cast<int32_t>(GetSelfPID()),
       base::Bind(&ServerChannelMojo::ClientChannelWasInitialized,
                  base::Unretained(this)));
 }
 
 void ServerChannelMojo::ClientChannelWasInitialized(int32_t peer_pid) {
   InitMessageReader(message_pipe_.Pass(), peer_pid);
 }
 
 void ServerChannelMojo::OnConnectionError() {
   listener()->OnChannelError();
 }
 
 void ServerChannelMojo::Close() {
   client_channel_.reset();
   message_pipe_.reset();
   ChannelMojo::Close();
 }
 
 #if defined(OS_POSIX) && !defined(OS_NACL)
 
 base::ScopedFD TakeOrDupFile(internal::PlatformFileAttachment* attachment) {
   return attachment->Owns() ? base::ScopedFD(attachment->TakePlatformFile())
                             : base::ScopedFD(dup(attachment->file()));
 }
 
 #endif
 
-} // namespace
+}  // namespace
 
 //------------------------------------------------------------------------------
 
+ChannelMojo::ChannelInfoDeleter::ChannelInfoDeleter(
+    scoped_refptr<base::TaskRunner> io_runner)
+    : io_runner(io_runner) {
+}
+
+ChannelMojo::ChannelInfoDeleter::~ChannelInfoDeleter() {
+}
+
 void ChannelMojo::ChannelInfoDeleter::operator()(
     mojo::embedder::ChannelInfo* ptr) const {
-  mojo::embedder::DestroyChannelOnIOThread(ptr);
+  if (base::ThreadTaskRunnerHandle::Get() == io_runner) {
+    mojo::embedder::DestroyChannelOnIOThread(ptr);
+  } else {
+    io_runner->PostTask(
+        FROM_HERE, base::Bind(&mojo::embedder::DestroyChannelOnIOThread, ptr));
+  }
 }
 
 //------------------------------------------------------------------------------
 
 // static
 bool ChannelMojo::ShouldBeUsed() {
   // TODO(morrita): Remove this if it sticks.
   return true;
 }
 
@@ skipping 37 matching lines @@
 
 ChannelMojo::ChannelMojo(ChannelMojo::Delegate* delegate,
                          scoped_refptr<base::TaskRunner> io_runner,
                          const ChannelHandle& handle,
                          Mode mode,
                          Listener* listener)
     : mode_(mode),
       listener_(listener),
       peer_pid_(base::kNullProcessId),
       io_runner_(io_runner),
+      channel_info_(nullptr, ChannelInfoDeleter(nullptr)),
       weak_factory_(this) {
   // Create MojoBootstrap after all members are set as it touches
   // ChannelMojo from a different thread.
   bootstrap_ = MojoBootstrap::Create(handle, mode, this);
   if (io_runner == base::MessageLoop::current()->message_loop_proxy()) {
     InitOnIOThread(delegate);
   } else {
     io_runner->PostTask(FROM_HERE,
                         base::Bind(&ChannelMojo::InitOnIOThread,
                                    base::Unretained(this), delegate));
   }
 }
 
 ChannelMojo::~ChannelMojo() {
   Close();
 }
 
 void ChannelMojo::InitOnIOThread(ChannelMojo::Delegate* delegate) {
   ipc_support_.reset(
       new ScopedIPCSupport(base::MessageLoop::current()->task_runner()));
   if (!delegate)
     return;
   delegate_ = delegate->ToWeakPtr();
   delegate_->OnChannelCreated(weak_factory_.GetWeakPtr());
 }
 
-mojo::ScopedMessagePipeHandle ChannelMojo::CreateMessagingPipe(
-    mojo::embedder::ScopedPlatformHandle handle) {
-  DCHECK(!channel_info_.get());
+void ChannelMojo::CreateMessagingPipe(
+    mojo::embedder::ScopedPlatformHandle handle,
+    const CreateMessagingPipeCallback& callback) {
+  auto return_callback = base::Bind(&ChannelMojo::OnMessagingPipeCreated,
+                                    weak_factory_.GetWeakPtr(), callback);
+  if (base::ThreadTaskRunnerHandle::Get() == io_runner_) {
+    CreateMessagingPipeOnIOThread(
+        handle.Pass(), base::ThreadTaskRunnerHandle::Get(), return_callback);
+  } else {
+    io_runner_->PostTask(
+        FROM_HERE,
+        base::Bind(&ChannelMojo::CreateMessagingPipeOnIOThread,
+                   base::Passed(&handle), base::ThreadTaskRunnerHandle::Get(),
+                   return_callback));
+  }
+}
+
+// static
+void ChannelMojo::CreateMessagingPipeOnIOThread(
+    mojo::embedder::ScopedPlatformHandle handle,
+    scoped_refptr<base::TaskRunner> callback_runner,
+    const CreateMessagingPipeOnIOThreadCallback& callback) {
   mojo::embedder::ChannelInfo* channel_info;
   mojo::ScopedMessagePipeHandle pipe =
       mojo::embedder::CreateChannelOnIOThread(handle.Pass(), &channel_info);
-  channel_info_.reset(channel_info);
-  return pipe.Pass();
+  if (base::ThreadTaskRunnerHandle::Get() == callback_runner) {
+    callback.Run(pipe.Pass(), channel_info);
+  } else {
+    callback_runner->PostTask(
+        FROM_HERE, base::Bind(callback, base::Passed(&pipe), channel_info));
+  }
+}
+
+void ChannelMojo::OnMessagingPipeCreated(
+    const CreateMessagingPipeCallback& callback,
+    mojo::ScopedMessagePipeHandle handle,
+    mojo::embedder::ChannelInfo* channel_info) {
+  DCHECK(!channel_info_.get());
+  channel_info_ = scoped_ptr<mojo::embedder::ChannelInfo, ChannelInfoDeleter>(
+      channel_info, ChannelInfoDeleter(io_runner_));
+  callback.Run(handle.Pass());
 }
 
 bool ChannelMojo::Connect() {
   DCHECK(!message_reader_);
   return bootstrap_->Connect();
 }
 
 void ChannelMojo::Close() {
   scoped_ptr<internal::MessagePipeReader, ReaderDeleter> to_be_deleted;
 
@@ skipping 183 matching lines @@
     if (!ok) {
       LOG(ERROR) << "Failed to add new Mojo handle.";
       return MOJO_RESULT_UNKNOWN;
     }
   }
 
   return MOJO_RESULT_OK;
 }
 
 }  // namespace IPC