Support early associated interface binding on ChannelMojo

ipc/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/ipc_channel_mojo.h"
 
 #include <stddef.h>
 #include <stdint.h>
 
 #include <memory>
 #include <utility>
 
 #include "base/bind.h"
 #include "base/bind_helpers.h"
 #include "base/command_line.h"
 #include "base/lazy_instance.h"
 #include "base/macros.h"
 #include "base/memory/ptr_util.h"
+#include "base/process/process_handle.h"
 #include "base/threading/thread_task_runner_handle.h"
 #include "build/build_config.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/ipc_mojo_bootstrap.h"
 #include "ipc/ipc_mojo_handle_attachment.h"
 #include "mojo/public/cpp/bindings/binding.h"
 #include "mojo/public/cpp/system/platform_handle.h"
@@ skipping 233 matching lines @@
     const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner) {
   return base::WrapUnique(new MojoChannelFactory(
       std::move(handle), Channel::MODE_CLIENT, ipc_task_runner));
 }
 
 ChannelMojo::ChannelMojo(
     mojo::ScopedMessagePipeHandle handle,
     Mode mode,
     Listener* listener,
     const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner)
-    : pipe_(handle.get()),
-      listener_(listener),
-      waiting_connect_(true),
-      weak_factory_(this) {
+    : pipe_(handle.get()), listener_(listener), weak_factory_(this) {
   // Create MojoBootstrap after all members are set as it touches
   // ChannelMojo from a different thread.
-  bootstrap_ = MojoBootstrap::Create(std::move(handle), mode, this);
+  bootstrap_ =
+      MojoBootstrap::Create(std::move(handle), mode, this, ipc_task_runner);
 }
 
 ChannelMojo::~ChannelMojo() {
   Close();
 }
 
 bool ChannelMojo::Connect() {
   WillConnect();
   {
     base::AutoLock lock(lock_);
     DCHECK(!task_runner_);
     task_runner_ = base::ThreadTaskRunnerHandle::Get();
     DCHECK(!message_reader_);
   }
   bootstrap_->Connect();
   return true;
 }
 
 void ChannelMojo::Close() {
   std::unique_ptr<internal::MessagePipeReader, ReaderDeleter> reader;
   {
     base::AutoLock lock(lock_);
+    associated_interfaces_.clear();

[yzshen1, 2016/07/20 17:13:15]

A few questions about lock usage:

- Is it safe to use |lock_| instead of |associated_interface_lock_| to guard
this clear call?

- Is the only reason to put |message_reader_| under |lock_| because it can be
set to null on the IO thread while on another thread someone tries to use it to
Send? If that is the case, can we set message_reader_ to close state without
deleting it so that we can remove the lock?

[Ken Rockot(use gerrit already), 2016/07/20 21:08:39]

Good catch :)
The only reason was because of thread-safe Send. Since we are not intending to
support that anymore (and ChannelMojo is marked as not having a thread-safe
Send) this is actually unnecessary. |message_reader_| is only accessed on the IO
thread now, so I've removed |lock_|.

     if (!message_reader_)
       return;
     // The reader's destructor may re-enter Close, so we swap it out first to
     // avoid deadlock when freeing it below.
     std::swap(message_reader_, reader);
-
-    // We might Close() before we Connect().
-    waiting_connect_ = false;
   }
 
   reader.reset();
 }
 
 // MojoBootstrap::Delegate implementation
-void ChannelMojo::OnPipesAvailable(
-    mojom::ChannelAssociatedPtrInfo send_channel,
-    mojom::ChannelAssociatedRequest receive_channel,
-    int32_t peer_pid) {
-  InitMessageReader(std::move(send_channel), std::move(receive_channel),
-                    peer_pid);
-}
+void ChannelMojo::OnPipesAvailable(mojom::ChannelAssociatedPtr sender,
+                                   mojom::ChannelAssociatedRequest receiver) {
+  sender->SetPeerPid(GetSelfPID());
 
-void ChannelMojo::OnBootstrapError() {
-  listener_->OnChannelError();
-}
-
-void ChannelMojo::OnAssociatedInterfaceRequest(
-    const std::string& name,
-    mojo::ScopedInterfaceEndpointHandle handle) {
-  auto iter = associated_interfaces_.find(name);
-  if (iter != associated_interfaces_.end())
-    iter->second.Run(std::move(handle));
-}
-
-void ChannelMojo::InitMessageReader(mojom::ChannelAssociatedPtrInfo sender,
-                                    mojom::ChannelAssociatedRequest receiver,
-                                    base::ProcessId peer_pid) {
-  mojom::ChannelAssociatedPtr sender_ptr;
-  sender_ptr.Bind(std::move(sender));
-  std::unique_ptr<internal::MessagePipeReader, ChannelMojo::ReaderDeleter>
-      reader(new internal::MessagePipeReader(
-          pipe_, std::move(sender_ptr), std::move(receiver), peer_pid, this));
-
-  bool connected = true;
   {
     base::AutoLock lock(lock_);
-    for (size_t i = 0; i < pending_messages_.size(); ++i) {
-      if (!reader->Send(std::move(pending_messages_[i]))) {
-        LOG(ERROR) << "Failed to flush pending messages";
-        pending_messages_.clear();
-        connected = false;
-        break;
-      }
-    }
-
-    if (connected) {
-      // We set |message_reader_| here and won't get any |pending_messages_|
-      // hereafter. Although we might have some if there is an error, we don't
-      // care. They cannot be sent anyway.
-      message_reader_ = std::move(reader);
-      pending_messages_.clear();
-      waiting_connect_ = false;
-    }
+    message_reader_.reset(new internal::MessagePipeReader(
+        pipe_, std::move(sender), std::move(receiver), this));
   }
-
-  if (connected)
-    listener_->OnChannelConnected(static_cast<int32_t>(GetPeerPID()));
-  else
-    OnPipeError();
 }
 
 void ChannelMojo::OnPipeError() {
   DCHECK(task_runner_);
   if (task_runner_->RunsTasksOnCurrentThread()) {
     listener_->OnChannelError();
   } else {
     task_runner_->PostTask(
         FROM_HERE,
         base::Bind(&ChannelMojo::OnPipeError, weak_factory_.GetWeakPtr()));
   }
 }
 
+void ChannelMojo::OnAssociatedInterfaceRequest(
+    const std::string& name,
+    mojo::ScopedInterfaceEndpointHandle handle) {
+  GenericAssociatedInterfaceFactory factory;
+  {
+    base::AutoLock locker(associated_interface_lock_);
+    auto iter = associated_interfaces_.find(name);
+    if (iter != associated_interfaces_.end())
+      factory = iter->second;
+  }
+
+  if (!factory.is_null())
+    factory.Run(std::move(handle));
+}
+
 bool ChannelMojo::Send(Message* message) {
+  std::unique_ptr<Message> scoped_message = base::WrapUnique(message);
   base::AutoLock lock(lock_);
-  if (!message_reader_) {
-    pending_messages_.push_back(base::WrapUnique(message));
-    // Counts as OK before the connection is established, but it's an
-    // error otherwise.
-    return waiting_connect_;
-  }
+  if (!message_reader_)
+    return false;
 
   // Comment copied from ipc_channel_posix.cc:
   // We can't close the pipe here, because calling OnChannelError may destroy
   // this object, and that would be bad if we are called from Send(). Instead,
   // we return false and hope the caller will close the pipe. If they do not,
   // the pipe will still be closed next time OnFileCanReadWithoutBlocking is
   // called.
   //
   // With Mojo, there's no OnFileCanReadWithoutBlocking, but we expect the
   // pipe's connection error handler will be invoked in its place.
-  return message_reader_->Send(base::WrapUnique(message));
+  return message_reader_->Send(std::move(scoped_message));
 }
 
 bool ChannelMojo::IsSendThreadSafe() const {
   return false;
 }
 
 base::ProcessId ChannelMojo::GetPeerPID() const {
   base::AutoLock lock(lock_);
   if (!message_reader_)
     return base::kNullProcessId;
 
   return message_reader_->GetPeerPid();
 }
 
 base::ProcessId ChannelMojo::GetSelfPID() const {
-  return bootstrap_->GetSelfPID();
+#if defined(OS_LINUX)
+  if (int global_pid = GetGlobalPid())
+    return global_pid;
+#endif  // OS_LINUX
+#if defined(OS_NACL)
+  return -1;
+#else
+  return base::GetCurrentProcId();
+#endif  // defined(OS_NACL)
 }
 
 Channel::AssociatedInterfaceSupport*
 ChannelMojo::GetAssociatedInterfaceSupport() { return this; }
 
+void ChannelMojo::OnPeerPidReceived() {
+  listener_->OnChannelConnected(static_cast<int32_t>(GetPeerPID()));
+}
+
 void ChannelMojo::OnMessageReceived(const Message& message) {
   TRACE_EVENT2("ipc,toplevel", "ChannelMojo::OnMessageReceived",
                "class", IPC_MESSAGE_ID_CLASS(message.type()),
                "line", IPC_MESSAGE_ID_LINE(message.type()));
   if (AttachmentBroker* broker = AttachmentBroker::GetGlobal()) {
     if (broker->OnMessageReceived(message))
       return;
   }
   listener_->OnMessageReceived(message);
   if (message.dispatch_error())
@@ skipping 64 matching lines @@
 }
 
 mojo::AssociatedGroup* ChannelMojo::GetAssociatedGroup() {
   DCHECK(bootstrap_);
   return bootstrap_->GetAssociatedGroup();
 }
 
 void ChannelMojo::AddGenericAssociatedInterface(
     const std::string& name,
     const GenericAssociatedInterfaceFactory& factory) {
+  base::AutoLock locker(associated_interface_lock_);
   auto result = associated_interfaces_.insert({ name, factory });
   DCHECK(result.second);
 }
 
 void ChannelMojo::GetGenericRemoteAssociatedInterface(
     const std::string& name,
     mojo::ScopedInterfaceEndpointHandle handle) {
-  DCHECK(message_reader_);
-  message_reader_->GetRemoteInterface(name, std::move(handle));
-}
-
-void ChannelMojo::SetProxyTaskRunner(
-    scoped_refptr<base::SingleThreadTaskRunner> task_runner) {
-  DCHECK(bootstrap_);
-  bootstrap_->SetProxyTaskRunner(task_runner);
+  base::AutoLock locker(lock_);
+  if (message_reader_)
+    message_reader_->GetRemoteInterface(name, std::move(handle));
 }
 
 }  // namespace IPC