Let IPC::SyncMessageFilter take advantage of thread-safe Send.

SyncMessageFilter is unnecessarily hopping to the IO thread
before writing to its underlying Sender when the underlying
Sender has a thread-safe Send implementation.

This fixes that.

BUG=516464

Review URL: https://codereview.chromium.org/1262253004

Cr-Commit-Position: refs/heads/master@{#342027}

15 files changed, 71 insertions, 32 deletions

diff --git a/components/nacl/loader/nacl_listener.cc b/components/nacl/loader/nacl_listener.cc
index defd931..9bbb6d9 100644
--- a/components/nacl/loader/nacl_listener.cc
+++ b/components/nacl/loader/nacl_listener.cc
@@ -262,8 +262,7 @@ void NaClListener::Listen() {
           switches::kProcessChannelID);
   channel_ = IPC::SyncChannel::Create(this, io_thread_.task_runner().get(),
                                       &shutdown_event_);
-  filter_ = new IPC::SyncMessageFilter(&shutdown_event_);
-  channel_->AddFilter(filter_.get());
+  filter_ = channel_->CreateSyncMessageFilter();
   channel_->AddFilter(new FileTokenMessageFilter());
   channel_->Init(channel_name, IPC::Channel::MODE_CLIENT, true);
   main_loop_ = base::MessageLoop::current();
diff --git a/content/child/child_thread_impl.cc b/content/child/child_thread_impl.cc
index b0b7dd4..3978af0 100644
--- a/content/child/child_thread_impl.cc
+++ b/content/child/child_thread_impl.cc
@@ -366,8 +366,7 @@ void ChildThreadImpl::Init(const Options& options) {
 
   mojo_application_.reset(new MojoApplication(GetIOTaskRunner()));
 
-  sync_message_filter_ =
-      new IPC::SyncMessageFilter(ChildProcess::current()->GetShutDownEvent());
+  sync_message_filter_ = channel_->CreateSyncMessageFilter();
   thread_safe_sender_ = new ThreadSafeSender(
       message_loop_->task_runner(), sync_message_filter_.get());
 
@@ -394,7 +393,6 @@ void ChildThreadImpl::Init(const Options& options) {
   push_dispatcher_ = new PushDispatcher(thread_safe_sender_.get());
 
   channel_->AddFilter(histogram_message_filter_.get());
-  channel_->AddFilter(sync_message_filter_.get());
   channel_->AddFilter(resource_message_filter_.get());
   channel_->AddFilter(quota_message_filter_->GetFilter());
   channel_->AddFilter(notification_dispatcher_->GetFilter());
diff --git a/content/child/indexed_db/indexed_db_dispatcher_unittest.cc b/content/child/indexed_db/indexed_db_dispatcher_unittest.cc
index 7e146e7..6e72d7bf 100644
--- a/content/child/indexed_db/indexed_db_dispatcher_unittest.cc
+++ b/content/child/indexed_db/indexed_db_dispatcher_unittest.cc
@@ -60,22 +60,27 @@ class MockDispatcher : public IndexedDBDispatcher {
   DISALLOW_COPY_AND_ASSIGN(MockDispatcher);
 };
 
+class MockSyncMessageFilter : public IPC::SyncMessageFilter {
+ public:
+  MockSyncMessageFilter()
+      : SyncMessageFilter(nullptr, false /* is_channel_send_thread_safe */) {}
+
+ private:
+  ~MockSyncMessageFilter() override {}
+};
+
 }  // namespace
 
 class IndexedDBDispatcherTest : public testing::Test {
  public:
   IndexedDBDispatcherTest()
-      : task_runner_(base::ThreadTaskRunnerHandle::Get()),
-        sync_message_filter_(new IPC::SyncMessageFilter(NULL)),
-        thread_safe_sender_(new ThreadSafeSender(task_runner_.get(),
-                                                 sync_message_filter_.get())) {}
+      : thread_safe_sender_(new ThreadSafeSender(
+            base::ThreadTaskRunnerHandle::Get(), new MockSyncMessageFilter)) {}
 
   void TearDown() override { blink::WebHeap::collectAllGarbageForTesting(); }
 
  protected:
   base::MessageLoop message_loop_;
-  scoped_refptr<base::SingleThreadTaskRunner> task_runner_;
-  scoped_refptr<IPC::SyncMessageFilter> sync_message_filter_;
   scoped_refptr<ThreadSafeSender> thread_safe_sender_;
 
  private:
diff --git a/content/child/indexed_db/webidbcursor_impl_unittest.cc b/content/child/indexed_db/webidbcursor_impl_unittest.cc
index a21babd..69fa803 100644
--- a/content/child/indexed_db/webidbcursor_impl_unittest.cc
+++ b/content/child/indexed_db/webidbcursor_impl_unittest.cc
@@ -115,15 +115,23 @@ class MockContinueCallbacks : public WebIDBCallbacks {
   WebVector<WebBlobInfo>* web_blob_info_;
 };
 
+class MockSyncMessageFilter : public IPC::SyncMessageFilter {
+ public:
+  MockSyncMessageFilter()
+      : SyncMessageFilter(nullptr, false /* is_channel_send_thread_safe */) {}
+
+ private:
+  ~MockSyncMessageFilter() override {}
+};
+
 }  // namespace
 
 class WebIDBCursorImplTest : public testing::Test {
  public:
   WebIDBCursorImplTest() {
     null_key_.assignNull();
-    sync_message_filter_ = new IPC::SyncMessageFilter(NULL);
     thread_safe_sender_ = new ThreadSafeSender(
-        base::ThreadTaskRunnerHandle::Get(), sync_message_filter_.get());
+        base::ThreadTaskRunnerHandle::Get(), new MockSyncMessageFilter);
     dispatcher_ =
         make_scoped_ptr(new MockDispatcher(thread_safe_sender_.get()));
   }
@@ -135,8 +143,6 @@ class WebIDBCursorImplTest : public testing::Test {
   scoped_ptr<MockDispatcher> dispatcher_;
 
  private:
-  scoped_refptr<IPC::SyncMessageFilter> sync_message_filter_;
-
   DISALLOW_COPY_AND_ASSIGN(WebIDBCursorImplTest);
 };
 
diff --git a/content/common/gpu/client/gpu_channel_host.cc b/content/common/gpu/client/gpu_channel_host.cc
index 4a41418..bf39359 100644
--- a/content/common/gpu/client/gpu_channel_host.cc
+++ b/content/common/gpu/client/gpu_channel_host.cc
@@ -78,9 +78,7 @@ void GpuChannelHost::Connect(const IPC::ChannelHandle& channel_handle,
       channel_handle, IPC::Channel::MODE_CLIENT, NULL, io_task_runner.get(),
       true, shutdown_event, factory_->GetAttachmentBroker());
 
-  sync_filter_ = new IPC::SyncMessageFilter(shutdown_event);
-
-  channel_->AddFilter(sync_filter_.get());
+  sync_filter_ = channel_->CreateSyncMessageFilter();
 
   channel_filter_ = new MessageFilter();
 
diff --git a/content/renderer/gpu/queue_message_swap_promise_unittest.cc b/content/renderer/gpu/queue_message_swap_promise_unittest.cc
index 1261cb2..76294fb 100644
--- a/content/renderer/gpu/queue_message_swap_promise_unittest.cc
+++ b/content/renderer/gpu/queue_message_swap_promise_unittest.cc
@@ -31,7 +31,7 @@ class TestRenderWidget : public RenderWidget {
 
 class TestSyncMessageFilter : public IPC::SyncMessageFilter {
  public:
-  TestSyncMessageFilter() : IPC::SyncMessageFilter(NULL) {}
+  TestSyncMessageFilter() : IPC::SyncMessageFilter(NULL, false) {}
 
   bool Send(IPC::Message* message) override {
     messages_.push_back(message);
diff --git a/ipc/ipc_channel_proxy.cc b/ipc/ipc_channel_proxy.cc
index 44fa42b..adb3d67 100644
--- a/ipc/ipc_channel_proxy.cc
+++ b/ipc/ipc_channel_proxy.cc
@@ -341,6 +341,10 @@ void ChannelProxy::Context::Send(Message* message) {
                             base::Passed(scoped_ptr<Message>(message))));
 }
 
+bool ChannelProxy::Context::IsChannelSendThreadSafe() const {
+  return channel_send_thread_safe_;
+}
+
 //-----------------------------------------------------------------------------
 
 // static
diff --git a/ipc/ipc_channel_proxy.h b/ipc/ipc_channel_proxy.h
index 6ff5cc2..f9f7d31 100644
--- a/ipc/ipc_channel_proxy.h
+++ b/ipc/ipc_channel_proxy.h
@@ -187,6 +187,9 @@ class IPC_EXPORT ChannelProxy : public Sender, public base::NonThreadSafe {
     // Sends |message| from appropriate thread.
     void Send(Message* message);
 
+    // Indicates if the underlying channel's Send is thread-safe.
+    bool IsChannelSendThreadSafe() const;
+
    protected:
     friend class base::RefCountedThreadSafe<Context>;
     ~Context() override;
diff --git a/ipc/ipc_sync_channel.cc b/ipc/ipc_sync_channel.cc
index 35b88a7..845eccd 100644
--- a/ipc/ipc_sync_channel.cc
+++ b/ipc/ipc_sync_channel.cc
@@ -460,6 +460,14 @@ void SyncChannel::SetRestrictDispatchChannelGroup(int group) {
   sync_context()->set_restrict_dispatch_group(group);
 }
 
+scoped_refptr<SyncMessageFilter> SyncChannel::CreateSyncMessageFilter() {
+  scoped_refptr<SyncMessageFilter> filter = new SyncMessageFilter(
+      sync_context()->shutdown_event(),
+      sync_context()->IsChannelSendThreadSafe());
+  AddFilter(filter.get());
+  return filter;
+}
+
 bool SyncChannel::Send(Message* message) {
 #ifdef IPC_MESSAGE_LOG_ENABLED
   std::string name;
diff --git a/ipc/ipc_sync_channel.h b/ipc/ipc_sync_channel.h
index 8b4b9a9..2fd1b11 100644
--- a/ipc/ipc_sync_channel.h
+++ b/ipc/ipc_sync_channel.h
@@ -15,6 +15,7 @@
 #include "ipc/ipc_channel_handle.h"
 #include "ipc/ipc_channel_proxy.h"
 #include "ipc/ipc_sync_message.h"
+#include "ipc/ipc_sync_message_filter.h"
 
 namespace base {
 class WaitableEvent;
@@ -116,6 +117,11 @@ class IPC_EXPORT SyncChannel : public ChannelProxy {
   // default) will be dispatched in any case.
   void SetRestrictDispatchChannelGroup(int group);
 
+  // Creates a new IPC::SyncMessageFilter and adds it to this SyncChannel.
+  // This should be used instead of directly constructing a new
+  // SyncMessageFilter.
+  scoped_refptr<IPC::SyncMessageFilter> CreateSyncMessageFilter();
+
  protected:
   class ReceivedSyncMsgQueue;
   friend class ReceivedSyncMsgQueue;
diff --git a/ipc/ipc_sync_channel_unittest.cc b/ipc/ipc_sync_channel_unittest.cc
index 314a4b8..44218fd 100644
--- a/ipc/ipc_sync_channel_unittest.cc
+++ b/ipc/ipc_sync_channel_unittest.cc
@@ -950,7 +950,7 @@ class TestSyncMessageFilter : public SyncMessageFilter {
       base::WaitableEvent* shutdown_event,
       Worker* worker,
       scoped_refptr<base::SingleThreadTaskRunner> task_runner)
-      : SyncMessageFilter(shutdown_event),
+      : SyncMessageFilter(shutdown_event, false),
         worker_(worker),
         task_runner_(task_runner) {}
 
diff --git a/ipc/ipc_sync_message_filter.cc b/ipc/ipc_sync_message_filter.cc
index d86835d..f6d485d 100644
--- a/ipc/ipc_sync_message_filter.cc
+++ b/ipc/ipc_sync_message_filter.cc
@@ -15,17 +15,14 @@
 
 namespace IPC {
 
-SyncMessageFilter::SyncMessageFilter(base::WaitableEvent* shutdown_event)
-    : sender_(NULL),
-      listener_task_runner_(base::ThreadTaskRunnerHandle::Get()),
-      shutdown_event_(shutdown_event) {
-}
-
 bool SyncMessageFilter::Send(Message* message) {
   if (!message->is_sync()) {
     {
       base::AutoLock auto_lock(lock_);
-      if (!io_task_runner_.get()) {
+      if (sender_ && is_channel_send_thread_safe_) {
+        sender_->Send(message);
+        return true;
+      } else if (!io_task_runner_.get()) {
         pending_messages_.push_back(message);
         return true;
       }
@@ -112,6 +109,14 @@ bool SyncMessageFilter::OnMessageReceived(const Message& message) {
   return false;
 }
 
+SyncMessageFilter::SyncMessageFilter(base::WaitableEvent* shutdown_event,
+                                     bool is_channel_send_thread_safe)
+    : sender_(NULL),
+      is_channel_send_thread_safe_(is_channel_send_thread_safe),
+      listener_task_runner_(base::ThreadTaskRunnerHandle::Get()),
+      shutdown_event_(shutdown_event) {
+}
+
 SyncMessageFilter::~SyncMessageFilter() {
 }
 
diff --git a/ipc/ipc_sync_message_filter.h b/ipc/ipc_sync_message_filter.h
index 32dffe3..2a3c4e8 100644
--- a/ipc/ipc_sync_message_filter.h
+++ b/ipc/ipc_sync_message_filter.h
@@ -21,6 +21,7 @@ class WaitableEvent;
 }
 
 namespace IPC {
+class SyncChannel;
 
 // This MessageFilter allows sending synchronous IPC messages from a thread
 // other than the listener thread associated with the SyncChannel.  It does not
@@ -29,8 +30,6 @@ namespace IPC {
 // be used to send simultaneous synchronous messages from different threads.
 class IPC_EXPORT SyncMessageFilter : public MessageFilter, public Sender {
  public:
-  explicit SyncMessageFilter(base::WaitableEvent* shutdown_event);
-
   // MessageSender implementation.
   bool Send(Message* message) override;
 
@@ -41,9 +40,14 @@ class IPC_EXPORT SyncMessageFilter : public MessageFilter, public Sender {
   bool OnMessageReceived(const Message& message) override;
 
  protected:
+  SyncMessageFilter(base::WaitableEvent* shutdown_event,
+                    bool is_channel_send_thread_safe);
+
   ~SyncMessageFilter() override;
 
  private:
+  friend class SyncChannel;
+
   void SendOnIOThread(Message* message);
   // Signal all the pending sends as done, used in an error condition.
   void SignalAllEvents();
@@ -51,6 +55,9 @@ class IPC_EXPORT SyncMessageFilter : public MessageFilter, public Sender {
   // The channel to which this filter was added.
   Sender* sender_;
 
+  // Indicates if |sender_|'s Send method is thread-safe.
+  bool is_channel_send_thread_safe_;
+
   // The process's main thread.
   scoped_refptr<base::SingleThreadTaskRunner> listener_task_runner_;
 
diff --git a/ppapi/nacl_irt/manifest_service.cc b/ppapi/nacl_irt/manifest_service.cc
index 8a6ec1f..41e8601 100644
--- a/ppapi/nacl_irt/manifest_service.cc
+++ b/ppapi/nacl_irt/manifest_service.cc
@@ -28,7 +28,8 @@ namespace ppapi {
 class ManifestMessageFilter : public IPC::SyncMessageFilter {
  public:
   ManifestMessageFilter(base::WaitableEvent* shutdown_event)
-      : SyncMessageFilter(shutdown_event),
+      : SyncMessageFilter(shutdown_event,
+                          false /* is_channel_send_thread_safe */),
         connected_event_(
             true /* manual_reset */, false /* initially_signaled */) {
   }
diff --git a/ppapi/proxy/plugin_dispatcher.cc b/ppapi/proxy/plugin_dispatcher.cc
index 15c521c..80fec1f 100644
--- a/ppapi/proxy/plugin_dispatcher.cc
+++ b/ppapi/proxy/plugin_dispatcher.cc
@@ -171,8 +171,7 @@ bool PluginDispatcher::InitPluginWithChannel(
   plugin_delegate_ = delegate;
   plugin_dispatcher_id_ = plugin_delegate_->Register(this);
 
-  sync_filter_ = new IPC::SyncMessageFilter(delegate->GetShutdownEvent());
-  channel()->AddFilter(sync_filter_.get());
+  sync_filter_ = channel()->CreateSyncMessageFilter();
 
   // The message filter will intercept and process certain messages directly
   // on the I/O thread.