Add support for Attachment Brokering of IPC::Channels on multiple threads.

Attachment brokering makes the assumption that there's a single thread
on which IO related to IPC::Channels occurs. This assumption is violated
when the flag --single-process is set.

BUG=590213

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

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

8 files changed, 74 insertions, 22 deletions

diff --git a/ipc/attachment_broker.cc b/ipc/attachment_broker.cc
index df3e724..f53a4cf 100644
--- a/ipc/attachment_broker.cc
+++ b/ipc/attachment_broker.cc
@@ -81,7 +81,9 @@ void AttachmentBroker::RemoveObserver(AttachmentBroker::Observer* observer) {
     observers_.erase(it);
 }
 
-void AttachmentBroker::RegisterCommunicationChannel(Endpoint* endpoint) {
+void AttachmentBroker::RegisterCommunicationChannel(
+    Endpoint* endpoint,
+    scoped_refptr<base::SingleThreadTaskRunner> runner) {
   NOTREACHED();
 }
 
diff --git a/ipc/attachment_broker.h b/ipc/attachment_broker.h
index 9bca157..70383ed 100644
--- a/ipc/attachment_broker.h
+++ b/ipc/attachment_broker.h
@@ -27,6 +27,7 @@
 
 namespace base {
 class SequencedTaskRunner;
+class SingleThreadTaskRunner;
 };
 
 namespace IPC {
@@ -94,7 +95,11 @@ class IPC_EXPORT AttachmentBroker : public Listener {
   // communicates attachment information with the broker process. In the broker
   // process, these channels must be registered and deregistered with the
   // Attachment Broker as they are created and destroyed.
-  virtual void RegisterCommunicationChannel(Endpoint* endpoint);
+  //
+  // Invocations of Send() on |endpoint| will occur on thread bound to |runner|.
+  virtual void RegisterCommunicationChannel(
+      Endpoint* endpoint,
+      scoped_refptr<base::SingleThreadTaskRunner> runner);
   virtual void DeregisterCommunicationChannel(Endpoint* endpoint);
 
   // In each unprivileged process, exactly one channel should be used to
diff --git a/ipc/attachment_broker_mac_unittest.cc b/ipc/attachment_broker_mac_unittest.cc
index 9442324..32d2833 100644
--- a/ipc/attachment_broker_mac_unittest.cc
+++ b/ipc/attachment_broker_mac_unittest.cc
@@ -563,7 +563,7 @@ int CommonPrivilegedProcessMain(OnMessageReceivedCallback callback,
 
   scoped_ptr<IPC::Channel> channel(IPC::Channel::CreateClient(
       IPCTestBase::GetChannelName(channel_name), &listener));
-  globals->broker->RegisterCommunicationChannel(channel.get());
+  globals->broker->RegisterCommunicationChannel(channel.get(), nullptr);
   CHECK(channel->Connect());
 
   globals->initial_resident_size = GetResidentSize();
diff --git a/ipc/attachment_broker_privileged.cc b/ipc/attachment_broker_privileged.cc
index 9006798..9f41892 100644
--- a/ipc/attachment_broker_privileged.cc
+++ b/ipc/attachment_broker_privileged.cc
@@ -6,7 +6,9 @@
 
 #include <algorithm>
 
+#include "base/bind.h"
 #include "base/lazy_instance.h"
+#include "base/location.h"
 #include "base/metrics/histogram_macros.h"
 #include "build/build_config.h"
 #include "ipc/ipc_endpoint.h"
@@ -110,18 +112,25 @@ void AttachmentBrokerPrivileged::CreateBrokerForSingleProcessTests() {
 }
 
 void AttachmentBrokerPrivileged::RegisterCommunicationChannel(
-    Endpoint* endpoint) {
+    Endpoint* endpoint,
+    scoped_refptr<base::SingleThreadTaskRunner> runner) {
   base::AutoLock auto_lock(*get_lock());
   endpoint->SetAttachmentBrokerEndpoint(true);
-  auto it = std::find(endpoints_.begin(), endpoints_.end(), endpoint);
+  auto it = std::find_if(endpoints_.begin(), endpoints_.end(),
+                         [endpoint](const EndpointRunnerPair& pair) {
+                           return pair.first == endpoint;
+                         });
   DCHECK(endpoints_.end() == it);
-  endpoints_.push_back(endpoint);
+  endpoints_.push_back(std::make_pair(endpoint, runner));
 }
 
 void AttachmentBrokerPrivileged::DeregisterCommunicationChannel(
     Endpoint* endpoint) {
   base::AutoLock auto_lock(*get_lock());
-  auto it = std::find(endpoints_.begin(), endpoints_.end(), endpoint);
+  auto it = std::find_if(endpoints_.begin(), endpoints_.end(),
+                         [endpoint](const EndpointRunnerPair& pair) {
+                           return pair.first == endpoint;
+                         });
   if (it != endpoints_.end())
     endpoints_.erase(it);
 }
@@ -130,15 +139,30 @@ bool AttachmentBrokerPrivileged::IsPrivilegedBroker() {
   return true;
 }
 
-Sender* AttachmentBrokerPrivileged::GetSenderWithProcessId(base::ProcessId id) {
+AttachmentBrokerPrivileged::EndpointRunnerPair
+AttachmentBrokerPrivileged::GetSenderWithProcessId(base::ProcessId id) {
   get_lock()->AssertAcquired();
   auto it = std::find_if(endpoints_.begin(), endpoints_.end(),
-                         [id](Endpoint* c) { return c->GetPeerPID() == id; });
+                         [id](const EndpointRunnerPair& pair) {
+                           return pair.first->GetPeerPID() == id;
+                         });
   if (it == endpoints_.end())
-    return nullptr;
+    return std::make_pair(nullptr, nullptr);
   return *it;
 }
 
+void AttachmentBrokerPrivileged::SendMessageToEndpoint(EndpointRunnerPair pair,
+                                                       Message* message) {
+  if (!pair.second || pair.second->BelongsToCurrentThread()) {
+    pair.first->Send(message);
+  } else {
+    pair.second->PostTask(
+        FROM_HERE,
+        base::Bind(&AttachmentBrokerPrivileged::SendMessageToEndpoint,
+                   base::Unretained(this), pair, message));
+  }
+}
+
 void AttachmentBrokerPrivileged::LogError(UMAError error) {
   UMA_HISTOGRAM_ENUMERATION(
       "IPC.AttachmentBrokerPrivileged.BrokerAttachmentError", error, ERROR_MAX);
diff --git a/ipc/attachment_broker_privileged.h b/ipc/attachment_broker_privileged.h
index 8b85599..1fff37f 100644
--- a/ipc/attachment_broker_privileged.h
+++ b/ipc/attachment_broker_privileged.h
@@ -5,9 +5,11 @@
 #ifndef IPC_ATTACHMENT_BROKER_PRIVILEGED_H_
 #define IPC_ATTACHMENT_BROKER_PRIVILEGED_H_
 
+#include <utility>
 #include <vector>
 
 #include "base/macros.h"
+#include "base/memory/ref_counted.h"
 #include "base/memory/scoped_ptr.h"
 #include "build/build_config.h"
 #include "ipc/attachment_broker.h"
@@ -48,17 +50,26 @@ class IPC_EXPORT AttachmentBrokerPrivileged : public IPC::AttachmentBroker {
   static void CreateBrokerForSingleProcessTests();
 
   // AttachmentBroker overrides.
-  void RegisterCommunicationChannel(Endpoint* endpoint) override;
+  void RegisterCommunicationChannel(
+      Endpoint* endpoint,
+      scoped_refptr<base::SingleThreadTaskRunner> runner) override;
   void DeregisterCommunicationChannel(Endpoint* endpoint) override;
   bool IsPrivilegedBroker() override;
 
  protected:
+  using EndpointRunnerPair =
+      std::pair<Endpoint*, scoped_refptr<base::SingleThreadTaskRunner>>;
+
   // Returns the sender whose peer's process id is |id|.
   // Returns nullptr if no sender is found.
   // The lock returned by get_lock() must already be acquired before calling
   // this method. The return value is only guaranteed to be valid while the lock
   // is held.
-  Sender* GetSenderWithProcessId(base::ProcessId id);
+  EndpointRunnerPair GetSenderWithProcessId(base::ProcessId id);
+
+  // Sends a message to the endpoint, dispatching onto another thread if
+  // necessary.
+  void SendMessageToEndpoint(EndpointRunnerPair pair, Message* message);
 
   // Errors that can be reported by subclasses.
   // These match tools/metrics/histograms.xml.
@@ -105,7 +116,9 @@ class IPC_EXPORT AttachmentBrokerPrivileged : public IPC::AttachmentBroker {
   void LogError(UMAError error);
 
  private:
-  std::vector<Endpoint*> endpoints_;
+  // A vector of Endpoints, and the SingleThreadTaskRunner that should be used
+  // to invoke Send() on each Endpoint.
+  std::vector<EndpointRunnerPair> endpoints_;
   DISALLOW_COPY_AND_ASSIGN(AttachmentBrokerPrivileged);
 };
 
diff --git a/ipc/attachment_broker_privileged_mac.cc b/ipc/attachment_broker_privileged_mac.cc
index 6b53ce0..fdf4715 100644
--- a/ipc/attachment_broker_privileged_mac.cc
+++ b/ipc/attachment_broker_privileged_mac.cc
@@ -174,8 +174,9 @@ bool AttachmentBrokerPrivilegedMac::RouteWireFormatToAnother(
   // Another process is the destination.
   base::ProcessId dest = wire_format.destination_process;
   base::AutoLock auto_lock(*get_lock());
-  Sender* sender = GetSenderWithProcessId(dest);
-  if (!sender) {
+  AttachmentBrokerPrivileged::EndpointRunnerPair pair =
+      GetSenderWithProcessId(dest);
+  if (!pair.first) {
     // Assuming that this message was not sent from a malicious process, the
     // channel endpoint that would have received this message will block
     // forever.
@@ -186,7 +187,8 @@ bool AttachmentBrokerPrivilegedMac::RouteWireFormatToAnother(
   }
 
   LogError(DESTINATION_FOUND);
-  sender->Send(new AttachmentBrokerMsg_MachPortHasBeenDuplicated(wire_format));
+  SendMessageToEndpoint(
+      pair, new AttachmentBrokerMsg_MachPortHasBeenDuplicated(wire_format));
   return true;
 }
 
@@ -229,7 +231,9 @@ void AttachmentBrokerPrivilegedMac::SendPrecursorsForProcess(
 
   if (!to_self) {
     base::AutoLock auto_lock(*get_lock());
-    if (!GetSenderWithProcessId(pid)) {
+    AttachmentBrokerPrivileged::EndpointRunnerPair pair =
+        GetSenderWithProcessId(pid);
+    if (!pair.first) {
       // If there is no sender, then the destination process is no longer
       // running, or never existed to begin with.
       LogError(DESTINATION_NOT_FOUND);
@@ -319,7 +323,9 @@ void AttachmentBrokerPrivilegedMac::ProcessExtractorsForProcess(
 
   {
     base::AutoLock auto_lock(*get_lock());
-    if (!GetSenderWithProcessId(pid)) {
+    AttachmentBrokerPrivileged::EndpointRunnerPair pair =
+        GetSenderWithProcessId(pid);
+    if (!pair.first) {
       // If there is no sender, then the source process is no longer running.
       LogError(ERROR_SOURCE_NOT_FOUND);
       delete it->second;
diff --git a/ipc/attachment_broker_privileged_win.cc b/ipc/attachment_broker_privileged_win.cc
index 3ce3ec5..0647012 100644
--- a/ipc/attachment_broker_privileged_win.cc
+++ b/ipc/attachment_broker_privileged_win.cc
@@ -86,8 +86,9 @@ void AttachmentBrokerPrivilegedWin::RouteDuplicatedHandle(
   // Another process is the destination.
   base::ProcessId dest = wire_format.destination_process;
   base::AutoLock auto_lock(*get_lock());
-  Sender* sender = GetSenderWithProcessId(dest);
-  if (!sender) {
+  AttachmentBrokerPrivileged::EndpointRunnerPair pair =
+      GetSenderWithProcessId(dest);
+  if (!pair.first) {
     // Assuming that this message was not sent from a malicious process, the
     // channel endpoint that would have received this message will block
     // forever.
@@ -98,7 +99,8 @@ void AttachmentBrokerPrivilegedWin::RouteDuplicatedHandle(
   }
 
   LogError(DESTINATION_FOUND);
-  sender->Send(new AttachmentBrokerMsg_WinHandleHasBeenDuplicated(wire_format));
+  SendMessageToEndpoint(
+      pair, new AttachmentBrokerMsg_WinHandleHasBeenDuplicated(wire_format));
 }
 
 AttachmentBrokerPrivilegedWin::HandleWireFormat
diff --git a/ipc/attachment_broker_privileged_win_unittest.cc b/ipc/attachment_broker_privileged_win_unittest.cc
index 9f720c8..3ab9111 100644
--- a/ipc/attachment_broker_privileged_win_unittest.cc
+++ b/ipc/attachment_broker_privileged_win_unittest.cc
@@ -506,7 +506,7 @@ int CommonPrivilegedProcessMain(OnMessageReceivedCallback callback,
   IPC::AttachmentBrokerPrivilegedWin broker;
   scoped_ptr<IPC::Channel> channel(IPC::Channel::CreateClient(
       IPCTestBase::GetChannelName(channel_name), &listener));
-  broker.RegisterCommunicationChannel(channel.get());
+  broker.RegisterCommunicationChannel(channel.get(), nullptr);
   CHECK(channel->Connect());
 
   while (true) {