Fix race condition with multiplexed message pipes.

mojo/edk/system/child_broker.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/child_broker.h"
 
 #include "base/bind.h"
 #include "base/logging.h"
 #include "mojo/edk/embedder/embedder_internal.h"
 #include "mojo/edk/embedder/platform_channel_pair.h"
@@ skipping 120 matching lines @@
       client_handle = channel_pair.PassClientHandle();
 #endif
       in_process_pipes_channel1_ = new RoutedRawChannel(
           server_handle.Pass(),
           base::Bind(&ChildBroker::ChannelDestructed, base::Unretained(this)));
       in_process_pipes_channel2_ = new RoutedRawChannel(
           client_handle.Pass(),
           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;
   }
 
   data.type = CONNECT_MESSAGE_PIPE;
   scoped_ptr<MessageInTransit> message(new MessageInTransit(
       MessageInTransit::Type::MESSAGE, sizeof(data), &data));
   pending_connects_[pipe_id] = message_pipe;
   WriteAsyncMessage(message.Pass());
 }
@@ skipping 45 matching lines @@
         static_cast<const PeerPipeConnectedMessage*>(message_view.bytes());
 
     uint64_t pipe_id = message->pipe_id;
     uint64_t peer_pid = message->process_id;
 
     CHECK(pending_connects_.find(pipe_id) != pending_connects_.end());
     MessagePipeDispatcher* pipe = pending_connects_[pipe_id];
     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();
   }
 }
 
 void ChildBroker::OnError(Error error) {
   // The parent process shut down.
 }
 
@@ skipping 30 matching lines @@
         async_channel_queue_.GetMessage());
   }
 
   while (!pending_inprocess_connects_.empty()) {
     ConnectMessagePipe(pending_inprocess_connects_.begin()->first,
                        pending_inprocess_connects_.begin()->second);
     pending_inprocess_connects_.erase(pending_inprocess_connects_.begin());
   }
 }
 
+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,
                                        void* response,
                                        uint32_t response_size) {
   CHECK(parent_sync_channel_.is_valid());
 
   bool result = true;
   DWORD bytes_written = 0;
   // This will always write in one chunk per
@@ skipping 32 matching lines @@
   if (WriteAndReadResponse(&message, handles, response_size)) {
     server->reset(PlatformHandle(handles[0]));
     client->reset(PlatformHandle(handles[1]));
   }
 }
 
 #endif
 
 }  // namespace edk
 }  // namespace mojo