Fix race condition with multiplexed message pipes.

The problem was that if a MessagePipeDispatcher was connected to a RoutedRawChannel that had buffered messages to it. OnReadMessage would be called first which would tell the awakables that it's ready for read.  But if the awakable called ReadMessage before GotNonTransferableChannel was called, then they wouldn't get any messages because channel_ was null.

BUG=561803
TEST=out/Debug/mojo_runner mojo:example_main --use-mus-in-renderer http://www.slashdot.org and then ctrl+n works

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

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

4 files changed, 44 insertions, 30 deletions

diff --git a/mojo/edk/system/broker_state.cc b/mojo/edk/system/broker_state.cc
index 4fad5a9..739cccd 100644
--- a/mojo/edk/system/broker_state.cc
+++ b/mojo/edk/system/broker_state.cc
@@ -80,15 +80,9 @@ void BrokerState::ConnectMessagePipe(uint64_t pipe_id,
           base::Bind(&BrokerState::ChannelDestructed, base::Unretained(this)));
     }
 
-    connected_pipes_[pending_connects_[pipe_id]] = in_process_pipes_channel1_;
-    connected_pipes_[message_pipe] = in_process_pipes_channel2_;
-    in_process_pipes_channel1_->AddRoute(pipe_id, pending_connects_[pipe_id]);
-    in_process_pipes_channel2_->AddRoute(pipe_id, message_pipe);
-    pending_connects_[pipe_id]->GotNonTransferableChannel(
-        in_process_pipes_channel1_->channel());
-    message_pipe->GotNonTransferableChannel(
-        in_process_pipes_channel2_->channel());
-
+    AttachMessagePipe(pending_connects_[pipe_id], pipe_id,
+                     in_process_pipes_channel1_);
+    AttachMessagePipe(message_pipe, pipe_id, in_process_pipes_channel2_);
     pending_connects_.erase(pipe_id);
     return;
   }
@@ -96,11 +90,9 @@ void BrokerState::ConnectMessagePipe(uint64_t pipe_id,
   if (pending_child_connects_.find(pipe_id) != pending_child_connects_.end()) {
     // A child process has already tried to connect.
     ChildBrokerHost* child_host = pending_child_connects_[pipe_id];
-    child_host->channel()->AddRoute(pipe_id, message_pipe);
+    AttachMessagePipe(message_pipe, pipe_id, child_host->channel());
     child_host->ConnectMessagePipe(pipe_id, 0);
     pending_child_connects_.erase(pipe_id);
-    connected_pipes_[message_pipe] = child_host->channel();
-    message_pipe->GotNonTransferableChannel(child_host->channel()->channel());
     return;
   }
 
@@ -178,9 +170,7 @@ void BrokerState::HandleConnectMessagePipe(ChildBrokerHost* pipe_process,
   if (pending_connects_.find(pipe_id) != pending_connects_.end()) {
     // This parent process is the other side of the given pipe.
     MessagePipeDispatcher* pending_pipe = pending_connects_[pipe_id];
-    connected_pipes_[pending_pipe] = pipe_process->channel();
-    pipe_process->channel()->AddRoute(pipe_id, pending_pipe);
-    pending_pipe->GotNonTransferableChannel(pipe_process->channel()->channel());
+    AttachMessagePipe(pending_pipe, pipe_id, pipe_process->channel());
     pipe_process->ConnectMessagePipe(pipe_id, 0);
     pending_connects_.erase(pipe_id);
     return;
@@ -236,5 +226,17 @@ void BrokerState::EnsureProcessesConnected(base::ProcessId pid1,
 void BrokerState::ChannelDestructed(RoutedRawChannel* channel) {
 }
 
+void BrokerState::AttachMessagePipe(MessagePipeDispatcher* message_pipe,
+                                    uint64_t pipe_id,
+                                    RoutedRawChannel* raw_channel) {
+  connected_pipes_[message_pipe] = raw_channel;
+  // Note: we must call GotNonTransferableChannel before AddRoute because there
+  // could be race conditions if the pipe got queued messages in |AddRoute| but
+  // then when it's read it returns no messages because it doesn't have the
+  // channel yet.
+  message_pipe->GotNonTransferableChannel(raw_channel->channel());
+  raw_channel->AddRoute(pipe_id, message_pipe);
+}
+
 }  // namespace edk
 }  // namespace mojo
diff --git a/mojo/edk/system/broker_state.h b/mojo/edk/system/broker_state.h
index 0b017e8..497a2b8 100644
--- a/mojo/edk/system/broker_state.h
+++ b/mojo/edk/system/broker_state.h
@@ -67,6 +67,11 @@ class MOJO_SYSTEM_IMPL_EXPORT BrokerState : NON_EXPORTED_BASE(public Broker) {
   // Called on the IO thread.
   void ChannelDestructed(RoutedRawChannel* channel);
 
+  // Helper method to connect the given MessagePipe to the channel.
+  void AttachMessagePipe(MessagePipeDispatcher* message_pipe,
+                         uint64_t pipe_id,
+                         RoutedRawChannel* raw_channel);
+
 #if defined(OS_WIN)
   // Used in the parent (unsandboxed) process to hold a mapping between HANDLES
   // and tokens. When a child process wants to send a HANDLE to another process,
diff --git a/mojo/edk/system/child_broker.cc b/mojo/edk/system/child_broker.cc
index 2964d20e..d686e74 100644
--- a/mojo/edk/system/child_broker.cc
+++ b/mojo/edk/system/child_broker.cc
@@ -138,15 +138,9 @@ void ChildBroker::ConnectMessagePipe(uint64_t pipe_id,
           base::Bind(&ChildBroker::ChannelDestructed, base::Unretained(this)));
     }
 
-    connected_pipes_[pending_connects_[pipe_id]] = in_process_pipes_channel1_;
-    connected_pipes_[message_pipe] = in_process_pipes_channel2_;
-    in_process_pipes_channel1_->AddRoute(pipe_id, pending_connects_[pipe_id]);
-    in_process_pipes_channel2_->AddRoute(pipe_id, message_pipe);
-    pending_connects_[pipe_id]->GotNonTransferableChannel(
-        in_process_pipes_channel1_->channel());
-    message_pipe->GotNonTransferableChannel(
-        in_process_pipes_channel2_->channel());
-
+    AttachMessagePipe(pending_connects_[pipe_id], pipe_id,
+                      in_process_pipes_channel1_);
+    AttachMessagePipe(message_pipe, pipe_id, in_process_pipes_channel2_);
     pending_connects_.erase(pipe_id);
     return;
   }
@@ -212,17 +206,13 @@ void ChildBroker::OnReadMessage(
     pending_connects_.erase(pipe_id);
     if (peer_pid == 0) {
       // The other side is in the parent process.
-      connected_pipes_[pipe] = parent_async_channel_;
-      parent_async_channel_->AddRoute(pipe_id, pipe);
-      pipe->GotNonTransferableChannel(parent_async_channel_->channel());
+      AttachMessagePipe(pipe, pipe_id, parent_async_channel_);
     } else if (channels_.find(peer_pid) == channels_.end()) {
       // We saw the peer process die before we got the reply from the parent.
       pipe->OnError(ERROR_READ_SHUTDOWN);
     } else {
       CHECK(connected_pipes_.find(pipe) == connected_pipes_.end());
-      connected_pipes_[pipe] = channels_[peer_pid];
-      channels_[peer_pid]->AddRoute(pipe_id, pipe);
-      pipe->GotNonTransferableChannel(channels_[peer_pid]->channel());
+      AttachMessagePipe(pipe, pipe_id, channels_[peer_pid]);
     }
   } else {
     NOTREACHED();
@@ -273,6 +263,18 @@ void ChildBroker::InitAsyncChannel(
   }
 }
 
+void ChildBroker::AttachMessagePipe(MessagePipeDispatcher* message_pipe,
+                                    uint64_t pipe_id,
+                                    RoutedRawChannel* raw_channel) {
+  connected_pipes_[message_pipe] = raw_channel;
+  // Note: we must call GotNonTransferableChannel before AddRoute because there
+  // could be race conditions if the pipe got queued messages in |AddRoute| but
+  // then when it's read it returns no messages because it doesn't have the
+  // channel yet.
+  message_pipe->GotNonTransferableChannel(raw_channel->channel());
+  raw_channel->AddRoute(pipe_id, message_pipe);
+}
+
 #if defined(OS_WIN)
 
 bool ChildBroker::WriteAndReadResponse(BrokerMessage* message,
diff --git a/mojo/edk/system/child_broker.h b/mojo/edk/system/child_broker.h
index 38cef56..c2d09b3 100644
--- a/mojo/edk/system/child_broker.h
+++ b/mojo/edk/system/child_broker.h
@@ -70,6 +70,11 @@ class MOJO_SYSTEM_IMPL_EXPORT ChildBroker
   // Initializes |parent_async_channel_|.
   void InitAsyncChannel(ScopedPlatformHandle parent_async_channel_handle);
 
+  // Helper method to connect the given MessagePipe to the channel.
+  void AttachMessagePipe(MessagePipeDispatcher* message_pipe,
+                         uint64_t pipe_id,
+                         RoutedRawChannel* raw_channel);
+
 #if defined(OS_WIN)
   // Helper method to write the given message and read back the result.
   bool WriteAndReadResponse(BrokerMessage* message,