ipc: AttachmentBroker dispatches notifications onto an appropriate thread.

Previously, there were no guarantees about the thread on which an observer of
AttachmentBroker would receive its notifications. Now, the observer is
guaranteed to get a notification on the same thread in which it registered
itself as an observer. This also makes notification of observers asynchronous,
which avoids issues related to re-entrant functions.

BUG=550938

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

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

6 files changed, 98 insertions, 18 deletions

diff --git a/ipc/attachment_broker.cc b/ipc/attachment_broker.cc
index 148e0d5..50e3e2d9 100644
--- a/ipc/attachment_broker.cc
+++ b/ipc/attachment_broker.cc
@@ -6,6 +6,10 @@
 
 #include <algorithm>
 
+#include "base/bind.h"
+#include "base/location.h"
+#include "base/sequenced_task_runner.h"
+
 namespace {
 IPC::AttachmentBroker* g_attachment_broker = nullptr;
 }
@@ -25,7 +29,7 @@ AttachmentBroker* AttachmentBroker::GetGlobal() {
   return g_attachment_broker;
 }
 
-AttachmentBroker::AttachmentBroker() {}
+AttachmentBroker::AttachmentBroker() : last_unique_id_(0) {}
 AttachmentBroker::~AttachmentBroker() {}
 
 bool AttachmentBroker::GetAttachmentWithId(
@@ -43,16 +47,29 @@ bool AttachmentBroker::GetAttachmentWithId(
   return false;
 }
 
-void AttachmentBroker::AddObserver(AttachmentBroker::Observer* observer) {
+void AttachmentBroker::AddObserver(
+    AttachmentBroker::Observer* observer,
+    const scoped_refptr<base::SequencedTaskRunner>& runner) {
   base::AutoLock auto_lock(*get_lock());
-  auto it = std::find(observers_.begin(), observers_.end(), observer);
-  if (it == observers_.end())
-    observers_.push_back(observer);
+  auto it = std::find_if(observers_.begin(), observers_.end(),
+                      [observer](const ObserverInfo& info) {
+                        return info.observer == observer;
+                      });
+  if (it == observers_.end()) {
+    ObserverInfo info;
+    info.observer = observer;
+    info.runner = runner;
+    info.unique_id = ++last_unique_id_;
+    observers_.push_back(info);
+  }
 }
 
 void AttachmentBroker::RemoveObserver(AttachmentBroker::Observer* observer) {
   base::AutoLock auto_lock(*get_lock());
-  auto it = std::find(observers_.begin(), observers_.end(), observer);
+  auto it = std::find_if(observers_.begin(), observers_.end(),
+                      [observer](const ObserverInfo& info) {
+                        return info.observer == observer;
+                      });
   if (it != observers_.end())
     observers_.erase(it);
 }
@@ -76,16 +93,40 @@ void AttachmentBroker::HandleReceivedAttachment(
 
 void AttachmentBroker::NotifyObservers(
     const BrokerableAttachment::AttachmentId& id) {
-  // Make a copy of observers_ to avoid mutations during iteration.
-  std::vector<Observer*> observers;
+  base::AutoLock auto_lock(*get_lock());
+
+  // Dispatch notifications onto the appropriate task runners. This has two
+  // effects:
+  //   1. Ensures that the notification is posted from the right task runner.
+  //   2. Avoids any complications from re-entrant functions, since one of the
+  //   observers may be halfway through processing some messages.
+  for (const auto& info : observers_) {
+    info.runner->PostTask(
+        FROM_HERE, base::Bind(&AttachmentBroker::NotifyObserver,
+                              base::Unretained(this), info.unique_id, id));
+  }
+}
+
+void AttachmentBroker::NotifyObserver(
+    int unique_id,
+    const BrokerableAttachment::AttachmentId& id) {
+  Observer* observer = nullptr;
   {
+    // Check that the same observer is still registered.
     base::AutoLock auto_lock(*get_lock());
-    observers = observers_;
+    auto it = std::find_if(observers_.begin(), observers_.end(),
+                           [unique_id](const ObserverInfo& info) {
+                             return info.unique_id == unique_id;
+                           });
+    if (it == observers_.end())
+      return;
+    observer = it->observer;
   }
 
-  for (Observer* observer : observers) {
-    observer->ReceivedBrokerableAttachmentWithId(id);
-  }
+  observer->ReceivedBrokerableAttachmentWithId(id);
 }
 
+AttachmentBroker::ObserverInfo::ObserverInfo() {}
+AttachmentBroker::ObserverInfo::~ObserverInfo() {}
+
 }  // namespace IPC
diff --git a/ipc/attachment_broker.h b/ipc/attachment_broker.h
index 273d764..0668892 100644
--- a/ipc/attachment_broker.h
+++ b/ipc/attachment_broker.h
@@ -24,6 +24,10 @@
 #define USE_ATTACHMENT_BROKER 0
 #endif  // defined(OS_WIN)
 
+namespace base {
+class SequencedTaskRunner;
+};
+
 namespace IPC {
 
 class AttachmentBroker;
@@ -77,7 +81,10 @@ class IPC_EXPORT AttachmentBroker : public Listener {
                            scoped_refptr<BrokerableAttachment>* attachment);
 
   // Any given observer should only ever add itself once to the observer list.
-  void AddObserver(Observer* observer);
+  // Notifications to |observer| will be posted to |runner|.
+  // The |observer| is expected to call RemoveObserver() before being destroyed.
+  void AddObserver(Observer* observer,
+                   const scoped_refptr<base::SequencedTaskRunner>& runner);
   void RemoveObserver(Observer* observer);
 
   // These two methods should only be called by the broker process.
@@ -99,6 +106,11 @@ class IPC_EXPORT AttachmentBroker : public Listener {
   // Informs the observers that a new BrokerableAttachment has been received.
   void NotifyObservers(const BrokerableAttachment::AttachmentId& id);
 
+  // Informs the observer identified by |unique_id| that a new
+  // BrokerableAttachment has been received.
+  void NotifyObserver(int unique_id,
+                      const BrokerableAttachment::AttachmentId& id);
+
   // This method is exposed for testing only.
   AttachmentVector* get_attachments() { return &attachments_; }
 
@@ -119,7 +131,20 @@ class IPC_EXPORT AttachmentBroker : public Listener {
   // better performance.
   AttachmentVector attachments_;
 
-  std::vector<Observer*> observers_;
+  struct ObserverInfo {
+    ObserverInfo();
+    ~ObserverInfo();
+
+    Observer* observer;
+    int unique_id;
+
+    // Notifications must be dispatched onto |runner|.
+    scoped_refptr<base::SequencedTaskRunner> runner;
+  };
+  std::vector<ObserverInfo> observers_;
+
+  // This member holds the last id given to an ObserverInfo.
+  int last_unique_id_;
 
   // The AttachmentBroker can be accessed from any thread, so modifications to
   // internal state must be guarded by a lock.
diff --git a/ipc/attachment_broker_mac_unittest.cc b/ipc/attachment_broker_mac_unittest.cc
index e3062b5..3188640 100644
--- a/ipc/attachment_broker_mac_unittest.cc
+++ b/ipc/attachment_broker_mac_unittest.cc
@@ -395,7 +395,7 @@ class IPCAttachmentBrokerMacTest : public IPCTestBase {
     if (!broker_.get())
       SetBroker(new IPC::AttachmentBrokerUnprivilegedMac);
 
-    broker_->AddObserver(&observer_);
+    broker_->AddObserver(&observer_, task_runner());
     CreateChannel(&proxy_listener_);
     broker_->DesignateBrokerCommunicationChannel(channel());
     ASSERT_TRUE(ConnectChannel());
@@ -856,10 +856,16 @@ TEST_F(IPCAttachmentBrokerMacTest, SendSharedMemoryHandleToSelf) {
     IPC::Message* message =
         new TestSharedMemoryHandleMsg1(100, shared_memory->handle(), 200);
     sender()->Send(message);
+
+    // Wait until the child process has sent this process a message.
     base::MessageLoop::current()->Run();
 
+    // Wait for any asynchronous activity to complete.
+    base::MessageLoop::current()->RunUntilIdle();
+
     // Get the received attachment.
     IPC::BrokerableAttachment::AttachmentId* id = get_observer()->get_id();
+    ASSERT_TRUE(id);
     scoped_refptr<IPC::BrokerableAttachment> received_attachment;
     get_broker()->GetAttachmentWithId(*id, &received_attachment);
     ASSERT_NE(received_attachment.get(), nullptr);
@@ -898,7 +904,7 @@ TEST_F(IPCAttachmentBrokerMacTest, SendSharedMemoryHandleChannelProxy) {
   MachPreForkSetUp();
 
   SetBroker(new IPC::AttachmentBrokerUnprivilegedMac);
-  get_broker()->AddObserver(get_observer());
+  get_broker()->AddObserver(get_observer(), task_runner());
 
   scoped_ptr<base::Thread> thread(
       new base::Thread("ChannelProxyTestServerThread"));
diff --git a/ipc/attachment_broker_privileged_win_unittest.cc b/ipc/attachment_broker_privileged_win_unittest.cc
index b0353d6..b5e1e98 100644
--- a/ipc/attachment_broker_privileged_win_unittest.cc
+++ b/ipc/attachment_broker_privileged_win_unittest.cc
@@ -225,7 +225,7 @@ class IPCAttachmentBrokerPrivilegedWinTest : public IPCTestBase {
   void CommonSetUp() {
     if (!broker_.get())
       set_broker(new IPC::AttachmentBrokerUnprivilegedWin);
-    broker_->AddObserver(&observer_);
+    broker_->AddObserver(&observer_, task_runner());
     CreateChannel(&proxy_listener_);
     broker_->DesignateBrokerCommunicationChannel(channel());
     ASSERT_TRUE(ConnectChannel());
diff --git a/ipc/ipc_channel_reader.cc b/ipc/ipc_channel_reader.cc
index e0cb2bf..f03f002 100644
--- a/ipc/ipc_channel_reader.cc
+++ b/ipc/ipc_channel_reader.cc
@@ -6,6 +6,7 @@
 
 #include <algorithm>
 
+#include "base/message_loop/message_loop.h"
 #include "ipc/ipc_listener.h"
 #include "ipc/ipc_logging.h"
 #include "ipc/ipc_message.h"
@@ -311,7 +312,8 @@ void ChannelReader::ReceivedBrokerableAttachmentWithId(
 
 void ChannelReader::StartObservingAttachmentBroker() {
 #if USE_ATTACHMENT_BROKER
-  GetAttachmentBroker()->AddObserver(this);
+  GetAttachmentBroker()->AddObserver(
+      this, base::MessageLoopForIO::current()->task_runner());
 #endif  // USE_ATTACHMENT_BROKER
 }
 
diff --git a/ipc/ipc_channel_reader_unittest.cc b/ipc/ipc_channel_reader_unittest.cc
index f49c275..ee29072 100644
--- a/ipc/ipc_channel_reader_unittest.cc
+++ b/ipc/ipc_channel_reader_unittest.cc
@@ -7,6 +7,7 @@
 #include <limits>
 #include <set>
 
+#include "base/run_loop.h"
 #include "ipc/attachment_broker.h"
 #include "ipc/brokerable_attachment.h"
 #include "ipc/ipc_channel_reader.h"
@@ -151,6 +152,8 @@ TEST(ChannelReaderTest, AttachmentAlreadyBrokered) {
 }
 
 TEST(ChannelReaderTest, AttachmentNotYetBrokered) {
+  scoped_ptr<base::MessageLoop> message_loop(new base::MessageLoopForIO());
+
   MockAttachmentBroker broker;
   MockChannelReader reader;
   reader.set_broker(&broker);
@@ -166,6 +169,9 @@ TEST(ChannelReaderTest, AttachmentNotYetBrokered) {
   EXPECT_EQ(nullptr, reader.get_last_dispatched_message());
 
   broker.AddAttachment(attachment);
+  base::RunLoop run_loop;
+  run_loop.RunUntilIdle();
+
   EXPECT_EQ(m, reader.get_last_dispatched_message());
 }