Add a Mojo EDK for Chrome that uses one OS pipe per message pipe.

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/public/cpp/bindings/binding.h"
+
+#if defined(USE_CHROME_EDK)
+#include "mojo/edk/embedder/embedder.h"
+#else
 #include "third_party/mojo/src/mojo/edk/embedder/embedder.h"
-#include "third_party/mojo/src/mojo/public/cpp/bindings/binding.h"
+#endif
 
 #if defined(OS_POSIX) && !defined(OS_NACL)
 #include "ipc/ipc_platform_file_attachment_posix.h"
 #endif
 
+// TODO(jam): do more tests on using channel on same thread if it supports it (
+// i.e. with USE_CHROME_EDK and Windows). Also see message_pipe_dispatcher.cc
+bool g_use_channel_on_io = true;
+
 namespace IPC {
 
 namespace {
 
 class MojoChannelFactory : public ChannelFactory {
  public:
   MojoChannelFactory(scoped_refptr<base::TaskRunner> io_runner,
                      ChannelHandle channel_handle,
                      Channel::Mode mode,
                      AttachmentBroker* broker)
@@ skipping 13 matching lines @@
 
  private:
   scoped_refptr<base::TaskRunner> io_runner_;
   ChannelHandle channel_handle_;
   Channel::Mode mode_;
   AttachmentBroker* broker_;
 };
 
 //------------------------------------------------------------------------------
 
-class ClientChannelMojo : public ChannelMojo, public ClientChannel {
+class ClientChannelMojo
+    : public ChannelMojo
+#if !defined(USE_CHROME_EDK)
+      , public ClientChannel
+#endif
+{
  public:
   ClientChannelMojo(scoped_refptr<base::TaskRunner> io_runner,
                     const ChannelHandle& handle,
                     Listener* listener,
-                    AttachmentBroker* broker);
-  ~ClientChannelMojo() override;
+                    AttachmentBroker* broker)
+      : ChannelMojo(io_runner, handle, Channel::MODE_CLIENT, listener, broker),
+#if !defined(USE_CHROME_EDK)
+        binding_(this),
+#endif
+        weak_factory_(this) {
+  }
+  ~ClientChannelMojo() override {}
   // MojoBootstrap::Delegate implementation
-  void OnPipeAvailable(mojo::embedder::ScopedPlatformHandle handle) override;
 
+#if defined(USE_CHROME_EDK)
+  void OnPipeAvailable(mojo::edk::ScopedPlatformHandle handle,
+                       int32 peer_pid) override {
+    InitMessageReader(
+        mojo::edk::CreateMessagePipe(handle.Pass()), peer_pid);
+}
+#else
+  void OnPipeAvailable(mojo::embedder::ScopedPlatformHandle handle,
+                       int32 peer_pid) override {
+    CreateMessagingPipe(handle.Pass(), base::Bind(&ClientChannelMojo::BindPipe,
+                                                  weak_factory_.GetWeakPtr()));
+  }
+#endif
+
+#if !defined(USE_CHROME_EDK)
   // ClientChannel implementation
   void Init(
       mojo::ScopedMessagePipeHandle pipe,
       int32_t peer_pid,
-      const mojo::Callback<void(int32_t)>& callback) override;
+      const mojo::Callback<void(int32_t)>& callback) override {
+   InitMessageReader(pipe.Pass(), static_cast<base::ProcessId>(peer_pid));
+   callback.Run(GetSelfPID());
+  }
+#endif
 
  private:
-  void BindPipe(mojo::ScopedMessagePipeHandle handle);
-  void OnConnectionError();
+#if !defined(USE_CHROME_EDK)
+  void BindPipe(mojo::ScopedMessagePipeHandle handle) {
+   // binding_.Bind(handle.Pass());
+  }
+  void OnConnectionError() {
+    listener()->OnChannelError();
+  }
 
   mojo::Binding<ClientChannel> binding_;
+#endif
   base::WeakPtrFactory<ClientChannelMojo> weak_factory_;
 
   DISALLOW_COPY_AND_ASSIGN(ClientChannelMojo);
 };
 
-ClientChannelMojo::ClientChannelMojo(scoped_refptr<base::TaskRunner> io_runner,
-                                     const ChannelHandle& handle,
-                                     Listener* listener,
-                                     AttachmentBroker* broker)
-    : ChannelMojo(io_runner, handle, Channel::MODE_CLIENT, listener, broker),
-      binding_(this),
-      weak_factory_(this) {
-}
-
-ClientChannelMojo::~ClientChannelMojo() {
-}
-
-void ClientChannelMojo::OnPipeAvailable(
-    mojo::embedder::ScopedPlatformHandle handle) {
-  CreateMessagingPipe(handle.Pass(), base::Bind(&ClientChannelMojo::BindPipe,
-                                                weak_factory_.GetWeakPtr()));
-}
-
-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());
-}
-
-void ClientChannelMojo::OnConnectionError() {
-  listener()->OnChannelError();
-}
-
 //------------------------------------------------------------------------------
 
 class ServerChannelMojo : public ChannelMojo {
  public:
   ServerChannelMojo(scoped_refptr<base::TaskRunner> io_runner,
                     const ChannelHandle& handle,
                     Listener* listener,
-                    AttachmentBroker* broker);
-  ~ServerChannelMojo() override;
+                    AttachmentBroker* broker)
+      : ChannelMojo(io_runner, handle, Channel::MODE_SERVER, listener, broker),
+        weak_factory_(this) {
+  }
+  ~ServerChannelMojo() override {
+    Close();
+  }
 
   // MojoBootstrap::Delegate implementation
-  void OnPipeAvailable(mojo::embedder::ScopedPlatformHandle handle) override;
+#if defined(USE_CHROME_EDK)
+  void OnPipeAvailable(mojo::edk::ScopedPlatformHandle handle,
+                       int32 peer_pid) override {
+    message_pipe_ = mojo::edk::CreateMessagePipe(handle.Pass());
+    if (!message_pipe_.is_valid()) {
+      LOG(WARNING) << "mojo::CreateMessagePipe failed: ";
+      listener()->OnChannelError();
+      return;
+    }
+    InitMessageReader(message_pipe_.Pass(), peer_pid);
+  }
+#else
+  void OnPipeAvailable(mojo::embedder::ScopedPlatformHandle handle,
+                       int32 peer_pid) override {
+    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)));
+  }
+#endif
   // Channel override
-  void Close() override;
+  void Close() override {
+#if !defined(USE_CHROME_EDK)
+    client_channel_.reset();
+#endif
+    message_pipe_.reset();
+    ChannelMojo::Close();
+  }
 
  private:
+#if !defined(USE_CHROME_EDK)
   void InitClientChannel(mojo::ScopedMessagePipeHandle peer_handle,
-                         mojo::ScopedMessagePipeHandle handle);
-  void OnConnectionError();
+                         mojo::ScopedMessagePipeHandle handle) {
+   client_channel_.Bind(
+        mojo::InterfacePtrInfo<ClientChannel>(handle.Pass(), 0u));
+    client_channel_.set_connection_error_handler(base::Bind(
+        &ServerChannelMojo::OnConnectionError, base::Unretained(this)));
+    client_channel_->Init(
+        peer_handle.Pass(), static_cast<int32_t>(GetSelfPID()),
+        base::Bind(&ServerChannelMojo::ClientChannelWasInitialized,
+                   base::Unretained(this)));
+  }
+#endif
+  void OnConnectionError() {
+    listener()->OnChannelError();
+  }
 
+#if !defined(USE_CHROME_EDK)
   // ClientChannelClient implementation
-  void ClientChannelWasInitialized(int32_t peer_pid);
+  void ClientChannelWasInitialized(int32_t peer_pid) {
+    InitMessageReader(message_pipe_.Pass(), peer_pid);
+  }
 
   mojo::InterfacePtr<ClientChannel> client_channel_;
+#endif
   mojo::ScopedMessagePipeHandle message_pipe_;
   base::WeakPtrFactory<ServerChannelMojo> weak_factory_;
 
   DISALLOW_COPY_AND_ASSIGN(ServerChannelMojo);
 };
 
-ServerChannelMojo::ServerChannelMojo(scoped_refptr<base::TaskRunner> io_runner,
-                                     const ChannelHandle& handle,
-                                     Listener* listener,
-                                     AttachmentBroker* broker)
-    : ChannelMojo(io_runner, handle, Channel::MODE_SERVER, listener, broker),
-      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::Close() {
-  client_channel_.reset();
-  message_pipe_.reset();
-  ChannelMojo::Close();
-}
-
-void ServerChannelMojo::InitClientChannel(
-    mojo::ScopedMessagePipeHandle peer_handle,
-    mojo::ScopedMessagePipeHandle handle) {
-  client_channel_.Bind(
-      mojo::InterfacePtrInfo<ClientChannel>(handle.Pass(), 0u));
-  client_channel_.set_connection_error_handler(base::Bind(
-      &ServerChannelMojo::OnConnectionError, base::Unretained(this)));
-  client_channel_->Init(
-      peer_handle.Pass(), static_cast<int32_t>(GetSelfPID()),
-      base::Bind(&ServerChannelMojo::ClientChannelWasInitialized,
-                 base::Unretained(this)));
-}
-
-void ServerChannelMojo::OnConnectionError() {
-  listener()->OnChannelError();
-}
-
-void ServerChannelMojo::ClientChannelWasInitialized(int32_t peer_pid) {
-  InitMessageReader(message_pipe_.Pass(), peer_pid);
-}
-
 #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
 
+#if !defined(USE_CHROME_EDK)
 //------------------------------------------------------------------------------
 
 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 {
   if (base::ThreadTaskRunnerHandle::Get() == io_runner) {
     mojo::embedder::DestroyChannelOnIOThread(ptr);
   } else {
     io_runner->PostTask(
         FROM_HERE, base::Bind(&mojo::embedder::DestroyChannelOnIOThread, ptr));
   }
 }
-
+#endif
 //------------------------------------------------------------------------------
 
 // static
 bool ChannelMojo::ShouldBeUsed() {
   // TODO(rockot): Investigate performance bottlenecks and hopefully reenable
   // this at some point. http://crbug.com/500019
   return false;
 }
 
 // static
@@ skipping 35 matching lines @@
 }
 
 ChannelMojo::ChannelMojo(scoped_refptr<base::TaskRunner> io_runner,
                          const ChannelHandle& handle,
                          Mode mode,
                          Listener* listener,
                          AttachmentBroker* broker)
     : listener_(listener),
       peer_pid_(base::kNullProcessId),
       io_runner_(io_runner),
+#if !defined(USE_CHROME_EDK)
       channel_info_(nullptr, ChannelInfoDeleter(nullptr)),
+#endif
       waiting_connect_(true),
       weak_factory_(this) {
   // Create MojoBootstrap after all members are set as it touches
   // ChannelMojo from a different thread.
   bootstrap_ = MojoBootstrap::Create(handle, mode, this, broker);
-  if (io_runner == base::MessageLoop::current()->task_runner()) {
+  if (g_use_channel_on_io ||
+      io_runner == base::MessageLoop::current()->task_runner()) {
     InitOnIOThread();
   } else {
     io_runner->PostTask(FROM_HERE, base::Bind(&ChannelMojo::InitOnIOThread,
                                               base::Unretained(this)));
   }
 }
 
 ChannelMojo::~ChannelMojo() {
   Close();
 }
 
 void ChannelMojo::InitOnIOThread() {
   ipc_support_.reset(
       new ScopedIPCSupport(base::MessageLoop::current()->task_runner()));
 }
 
+#if !defined(USE_CHROME_EDK)
 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_) {
+  if (g_use_channel_on_io ||
+      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));
   }
 }
@@ skipping 16 matching lines @@
 
 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());
 }
+#endif
 
 bool ChannelMojo::Connect() {
   DCHECK(!message_reader_);
   return bootstrap_->Connect();
 }
 
 void ChannelMojo::Close() {
   scoped_ptr<internal::MessagePipeReader, ReaderDeleter> to_be_deleted;
 
   {
     // |message_reader_| has to be cleared inside the lock,
     // but the instance has to be deleted outside.
     base::AutoLock l(lock_);
     to_be_deleted = message_reader_.Pass();
     // We might Close() before we Connect().
     waiting_connect_ = false;
   }
 
+#if !defined(USE_CHROME_EDK)
   channel_info_.reset();
+#endif
   ipc_support_.reset();
   to_be_deleted.reset();
 }
 
 void ChannelMojo::OnBootstrapError() {
   listener_->OnChannelError();
 }
 
 namespace {
 
@@ skipping 113 matching lines @@
           base::ScopedFD file =
               TakeOrDupFile(static_cast<IPC::internal::PlatformFileAttachment*>(
                   attachment.get()));
           if (!file.is_valid()) {
             DPLOG(WARNING) << "Failed to dup FD to transmit.";
             set->CommitAll();
             return MOJO_RESULT_UNKNOWN;
           }
 
           MojoHandle wrapped_handle;
+#if defined(USE_CHROME_EDK)
+          MojoResult wrap_result = CreatePlatformHandleWrapper(
+              mojo::edk::ScopedPlatformHandle(
+                  mojo::edk::PlatformHandle(file.release())),
+              &wrapped_handle);
+#else
           MojoResult wrap_result = CreatePlatformHandleWrapper(
               mojo::embedder::ScopedPlatformHandle(
                   mojo::embedder::PlatformHandle(file.release())),
               &wrapped_handle);
+#endif
           if (MOJO_RESULT_OK != wrap_result) {
             LOG(WARNING) << "Pipe failed to wrap handles. Closing: "
                          << wrap_result;
             set->CommitAll();
             return wrap_result;
           }
 
           handles->push_back(wrapped_handle);
         }
 #else
@@ skipping 32 matching lines @@
     if (!ok) {
       LOG(ERROR) << "Failed to add new Mojo handle.";
       return MOJO_RESULT_UNKNOWN;
     }
   }
 
   return MOJO_RESULT_OK;
 }
 
 }  // namespace IPC