Fix race condition with multiplexed message pipes.

mojo/edk/system/broker_state.cc

 // Copyright 2015 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 "mojo/edk/system/broker_state.h"
 
 #include "base/bind.h"
 #include "base/rand_util.h"
 #include "mojo/edk/embedder/embedder_internal.h"
 #include "mojo/edk/embedder/platform_channel_pair.h"
@@ skipping 62 matching lines @@
     if (!in_process_pipes_channel1_) {
       PlatformChannelPair channel_pair;
       in_process_pipes_channel1_ = new RoutedRawChannel(
           channel_pair.PassServerHandle(),
           base::Bind(&BrokerState::ChannelDestructed, base::Unretained(this)));
       in_process_pipes_channel2_ = new RoutedRawChannel(
           channel_pair.PassClientHandle(),
           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;
   }
 
   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;
   }
 
   pending_connects_[pipe_id] = message_pipe;
 }
 
 void BrokerState::CloseMessagePipe(uint64_t pipe_id,
                                    MessagePipeDispatcher* message_pipe) {
   DCHECK(internal::g_io_thread_task_runner->RunsTasksOnCurrentThread());
   base::AutoLock auto_lock(lock_);
@@ skipping 57 matching lines @@
         pipe_id, pipe_process->GetProcessId());
     pipe_process->ConnectMessagePipe(
         pipe_id, pending_pipe_process->GetProcessId());
     pending_child_connects_.erase(pipe_id);
     return;
   }
 
   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;
   }
 
   // This is the first connection request for pipe_id to reach the parent.
   pending_child_connects_[pipe_id] = pipe_process;
 }
 
 void BrokerState::HandleCancelConnectMessagePipe(uint64_t pipe_id) {
@@ skipping 35 matching lines @@
   CHECK(child_processes_.find(pid2) != child_processes_.end());
   child_processes_[pid1]->ConnectToProcess(pid2,
                                            channel_pair.PassServerHandle());
   child_processes_[pid2]->ConnectToProcess(pid1,
                                            channel_pair.PassClientHandle());
 }
 
 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