ChannelProxy: Send() from UI thread when ChannelMojo is used.

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 "ipc/ipc_listener.h"
@@ skipping 157 matching lines @@
 void ChannelMojo::ChannelInfoDeleter::operator()(
     mojo::embedder::ChannelInfo* ptr) const {
   mojo::embedder::DestroyChannel(ptr);
 }
 
 //------------------------------------------------------------------------------
 
 // static
 bool ChannelMojo::ShouldBeUsed() {
   // TODO(morrita): Turn this on for a set of platforms.
-  return false;

[Hajime Morrita, 2014/12/05 02:54:09]

I do this just for exercising tests. No intention to do this in this CL.

+  return true;
 }
 
 // static
 scoped_ptr<ChannelMojo> ChannelMojo::Create(ChannelMojo::Delegate* delegate,
                                             const ChannelHandle& channel_handle,
                                             Mode mode,
                                             Listener* listener) {
   switch (mode) {
     case Channel::MODE_CLIENT:
       return make_scoped_ptr(
@@ skipping 44 matching lines @@
               &ChannelMojo::InitDelegate, base::Unretained(this), delegate));
     }
   }
 }
 
 ChannelMojo::~ChannelMojo() {
   Close();
 }
 
 void ChannelMojo::InitDelegate(ChannelMojo::Delegate* delegate) {
+  delegate->OnChannelCreated(weak_factory_.GetWeakPtr());
+  base::AutoLock l(lock_);
   delegate_ = delegate->ToWeakPtr();
-  delegate_->OnChannelCreated(weak_factory_.GetWeakPtr());
 }
 
 mojo::ScopedMessagePipeHandle ChannelMojo::CreateMessagingPipe(
     mojo::embedder::ScopedPlatformHandle handle) {
   DCHECK(!channel_info_.get());
   mojo::embedder::ChannelInfo* channel_info;
   mojo::ScopedMessagePipeHandle pipe =
       mojo::embedder::CreateChannelOnIOThread(handle.Pass(), &channel_info);
   channel_info_.reset(channel_info);
   return pipe.Pass();
 }
 
 bool ChannelMojo::Connect() {
   DCHECK(!message_reader_);
   return bootstrap_->Connect();
 }
 
 void ChannelMojo::Close() {
-  message_reader_.reset();
+  // |message_reader_| has to be cleared inside the lock,
+  // but the instance has to be deleted outside.
+  scoped_ptr<internal::MessagePipeReader, ReaderDeleter> to_be_dead;
+
+  base::AutoLock l(lock_);
+  to_be_dead = message_reader_.Pass();
   channel_info_.reset();
 }
 
 void ChannelMojo::OnBootstrapError() {
   listener_->OnChannelError();
 }
 
 void ChannelMojo::InitMessageReader(mojo::ScopedMessagePipeHandle pipe,
                                     int32_t peer_pid) {
-  message_reader_ =
+  scoped_ptr<internal::MessagePipeReader> reader =
       make_scoped_ptr(new internal::MessagePipeReader(pipe.Pass(), this));
+  MojoResult send_result = MOJO_RESULT_OK;
 
-  for (size_t i = 0; i < pending_messages_.size(); ++i) {
-    bool sent = message_reader_->Send(make_scoped_ptr(pending_messages_[i]));
-    pending_messages_[i] = NULL;
-    if (!sent) {
-      pending_messages_.clear();
-      listener_->OnChannelError();
-      return;
+  {
+    base::AutoLock l(lock_);
+    ScopedVector<Message> pending;
+
+    pending_messages_.swap(pending);
+    for (size_t i = 0; i < pending.size(); ++i) {
+      send_result = reader->Send(make_scoped_ptr(pending[i]));
+      pending[i] = nullptr;
+      if (send_result != MOJO_RESULT_OK)
+        break;
     }
+
+    // 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.
+    if (send_result == MOJO_RESULT_OK)
+      message_reader_ = reader.Pass();
   }
 
-  pending_messages_.clear();
+  // This should be done outside the |lock_| as delegates are called through
+  // |CloseWithError()|.
+  if (reader.get()) {
+    DCHECK(!message_reader_.get());
+    reader->CloseWithError(send_result);
+    return;
+  }
 
+  DCHECK(message_reader_.get());
   set_peer_pid(peer_pid);
   listener_->OnChannelConnected(static_cast<int32_t>(GetPeerPID()));
 }
 
 void ChannelMojo::OnPipeClosed(internal::MessagePipeReader* reader) {
   Close();
 }
 
 void ChannelMojo::OnPipeError(internal::MessagePipeReader* reader) {
   listener_->OnChannelError();
 }
 
+base::TaskRunner* ChannelMojo::GetIOTaskRunner() {
+  base::AutoLock l(lock_);
+  if (!delegate_)
+    return nullptr;
+  return delegate_->GetIOTaskRunner().get();
+}
 
+// Reminder: Be thread-safe.
 bool ChannelMojo::Send(Message* message) {
-  if (!message_reader_) {
-    pending_messages_.push_back(message);
-    return true;
+  MojoResult send_result = MOJO_RESULT_OK;
+
+  {
+    base::AutoLock l(lock_);
+
+    if (!message_reader_) {
+      pending_messages_.push_back(message);
+      return true;
+    }
+
+    send_result = message_reader_->Send(make_scoped_ptr(message));
   }
 
-  return message_reader_->Send(make_scoped_ptr(message));
+  if (send_result != MOJO_RESULT_OK) {
+    if (base::TaskRunner* runner = GetIOTaskRunner()) {
+      runner->PostTask(
+          FROM_HERE,
+          base::Bind(&internal::MessagePipeReader::CloseWithError,
+                     base::Unretained(message_reader_.get()), send_result));
+      // The error shouln't be visible at this point.
+      // It will be known eventually by |CloseWithError()| on IO thread.
+      return true;
+    }
+
+    message_reader_->CloseWithError(send_result);
+    return false;
+  }
+
+  return true;
+}
+
+bool ChannelMojo::IsSendThreadSafe() const {
+  return true;
 }
 
 base::ProcessId ChannelMojo::GetPeerPID() const {
   return peer_pid_;
 }
 
 base::ProcessId ChannelMojo::GetSelfPID() const {
   return base::GetCurrentProcId();
 }
 
@@ skipping 74 matching lines @@
 
     fdset->CommitAll();
   }
 
   return MOJO_RESULT_OK;
 }
 
 #endif  // defined(OS_POSIX) && !defined(OS_NACL)
 
 }  // namespace IPC