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

ipc/ipc_channel_proxy.cc

 // Copyright (c) 2012 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_proxy.h"
 
 #include "base/bind.h"
 #include "base/compiler_specific.h"
 #include "base/location.h"
 #include "base/memory/ref_counted.h"
@@ skipping 12 matching lines @@
 //------------------------------------------------------------------------------
 
 ChannelProxy::Context::Context(
     Listener* listener,
     const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner)
     : listener_task_runner_(base::ThreadTaskRunnerHandle::Get()),
       listener_(listener),
       ipc_task_runner_(ipc_task_runner),
       channel_connected_called_(false),
       message_filter_router_(new MessageFilterRouter()),
+      channel_send_thread_safe_(false),
       peer_pid_(base::kNullProcessId) {
   DCHECK(ipc_task_runner_.get());
   // The Listener thread where Messages are handled must be a separate thread
   // to avoid oversubscribing the IO thread. If you trigger this error, you
   // need to either:
   // 1) Create the ChannelProxy on a different thread, or
   // 2) Just use Channel
   // Note, we currently make an exception for a NULL listener. That usage
   // basically works, but is outside the intent of ChannelProxy. This support
   // will disappear, so please don't rely on it. See crbug.com/364241
   DCHECK(!listener || (ipc_task_runner_.get() != listener_task_runner_.get()));
 }
 
 ChannelProxy::Context::~Context() {
 }
 
 void ChannelProxy::Context::ClearIPCTaskRunner() {
   ipc_task_runner_ = NULL;
 }
 
 void ChannelProxy::Context::CreateChannel(scoped_ptr<ChannelFactory> factory) {
   DCHECK(!channel_);
   channel_id_ = factory->GetName();
   channel_ = factory->BuildChannel(this);
+  channel_send_thread_safe_ = channel_->IsSendThreadSafe();
 }
 
 bool ChannelProxy::Context::TryFilters(const Message& message) {
   DCHECK(message_filter_router_);
 #ifdef IPC_MESSAGE_LOG_ENABLED
   Logging* logger = Logging::GetInstance();
   if (logger->Enabled())
     logger->OnPreDispatchMessage(message);
 #endif
 
@@ skipping 81 matching lines @@
     filters_[i]->OnFilterRemoved();
   }
 
   // We don't need the filters anymore.
   message_filter_router_->Clear();
   filters_.clear();
   // We don't need the lock, because at this point, the listener thread can't
   // access it any more.
   pending_filters_.clear();
 
-  channel_.reset();
+  ClearChannel();
 
   // Balance with the reference taken during startup.  This may result in
   // self-destruction.
   Release();
 }
 
 void ChannelProxy::Context::Clear() {
   listener_ = NULL;
 }
 
@@ skipping 127 matching lines @@
   if (listener_)
     listener_->OnChannelError();
 }
 
 // Called on the listener's thread
 void ChannelProxy::Context::OnDispatchBadMessage(const Message& message) {
   if (listener_)
     listener_->OnBadMessageReceived(message);
 }
 
+void ChannelProxy::Context::ClearChannel() {
+  base::AutoLock l(channel_lifetime_lock_);
+  channel_.reset();
+}
+
+void ChannelProxy::Context::SendFromThisThread(Message* message) {
+  base::AutoLock l(channel_lifetime_lock_);
+  if (!channel_)
+    return;
+  DCHECK(channel_->IsSendThreadSafe());
+  channel_->Send(message);
+}
+
+void ChannelProxy::Context::Send(Message* message) {
+  if (channel_send_thread_safe_) {
+    SendFromThisThread(message);
+    return;
+  }
+
+  ipc_task_runner()->PostTask(
+      FROM_HERE, base::Bind(&ChannelProxy::Context::OnSendMessage, this,
+                            base::Passed(scoped_ptr<Message>(message))));
+}
+
 //-----------------------------------------------------------------------------
 
 // static
 scoped_ptr<ChannelProxy> ChannelProxy::Create(
     const IPC::ChannelHandle& channel_handle,
     Channel::Mode mode,
     Listener* listener,
     const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner) {
   scoped_ptr<ChannelProxy> channel(new ChannelProxy(listener, ipc_task_runner));
   channel->Init(channel_handle, mode, true);
@@ skipping 84 matching lines @@
 bool ChannelProxy::Send(Message* message) {
   DCHECK(did_init_);
 
   // TODO(alexeypa): add DCHECK(CalledOnValidThread()) here. Currently there are
   // tests that call Send() from a wrong thread. See http://crbug.com/163523.
 
 #ifdef IPC_MESSAGE_LOG_ENABLED
   Logging::GetInstance()->OnSendMessage(message, context_->channel_id());
 #endif
 
-  context_->ipc_task_runner()->PostTask(
-      FROM_HERE,
-      base::Bind(&ChannelProxy::Context::OnSendMessage,
-                 context_, base::Passed(scoped_ptr<Message>(message))));
+  context_->Send(message);
   return true;
 }
 
 void ChannelProxy::AddFilter(MessageFilter* filter) {
   DCHECK(CalledOnValidThread());
 
   context_->AddFilter(filter);
 }
 
 void ChannelProxy::RemoveFilter(MessageFilter* filter) {
@@ skipping 28 matching lines @@
   Channel* channel = context_.get()->channel_.get();
   // Channel must have been created first.
   DCHECK(channel) << context_.get()->channel_id_;
   return channel->TakeClientFileDescriptor();
 }
 #endif
 
 //-----------------------------------------------------------------------------
 
 }  // namespace IPC