Fix/reland 505555fa8031fc06d67e9ed43e3021d8d84cb44f (crrev.com/773983002).

mojo/edk/system/message_pipe.cc

 // Copyright 2013 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/message_pipe.h"
 
 #include "base/logging.h"
 #include "mojo/edk/system/channel.h"
 #include "mojo/edk/system/channel_endpoint.h"
 #include "mojo/edk/system/channel_endpoint_id.h"
@@ skipping 121 matching lines @@
 }
 
 // TODO(vtl): Handle flags.
 MojoResult MessagePipe::WriteMessage(
     unsigned port,
     UserPointer<const void> bytes,
     uint32_t num_bytes,
     std::vector<DispatcherTransport>* transports,
     MojoWriteMessageFlags flags) {
   DCHECK(port == 0 || port == 1);
-  return EnqueueMessage(
+
+  base::AutoLock locker(lock_);
+  return EnqueueMessageNoLock(
       GetPeerPort(port),
       make_scoped_ptr(new MessageInTransit(
           MessageInTransit::kTypeEndpoint,
           MessageInTransit::kSubtypeEndpointData, num_bytes, bytes)),
       transports);
 }
 
 MojoResult MessagePipe::ReadMessage(unsigned port,
                                     UserPointer<void> bytes,
                                     UserPointer<uint32_t> num_bytes,
@@ skipping 127 matching lines @@
   // Convert the local endpoint to a proxy endpoint (moving the message queue)
   // and attach it to the channel.
   s->receiver_endpoint_id =
       channel->AttachAndRunEndpoint(channel_endpoint, false);
   DVLOG(2) << "Serializing message pipe (remote ID = "
            << s->receiver_endpoint_id << ")";
   *actual_size = sizeof(SerializedMessagePipe);
   return true;
 }
 
-void MessagePipe::OnReadMessage(unsigned port,
-                                scoped_ptr<MessageInTransit> message) {
+bool MessagePipe::OnReadMessage(unsigned port, MessageInTransit* message) {
+  base::AutoLock locker(lock_);
+
+  if (!endpoints_[port]) {
+    // This will happen only on the rare occasion that the call to
+    // |OnReadMessage()| is racing with us calling
+    // |ChannelEndpoint::ReplaceClient()|, in which case we reject the message,
+    // and the |ChannelEndpoint| can retry (calling the new client's
+    // |OnReadMessage()|).
+    return false;
+  }
+
   // This is called when the |ChannelEndpoint| for the
   // |ProxyMessagePipeEndpoint| |port| receives a message (from the |Channel|).
   // We need to pass this message on to its peer port (typically a
   // |LocalMessagePipeEndpoint|).
-  MojoResult result =
-      EnqueueMessage(GetPeerPort(port), message.Pass(), nullptr);
+  MojoResult result = EnqueueMessageNoLock(GetPeerPort(port),
+                                           make_scoped_ptr(message), nullptr);
   DLOG_IF(WARNING, result != MOJO_RESULT_OK)
-      << "EnqueueMessage() failed (result  = " << result << ")";
+      << "EnqueueMessageNoLock() failed (result  = " << result << ")";
+  return true;
 }
 
 void MessagePipe::OnDetachFromChannel(unsigned port) {
   Close(port);
 }
 
 MessagePipe::MessagePipe() {
 }
 
 MessagePipe::~MessagePipe() {
   // Owned by the dispatchers. The owning dispatchers should only release us via
   // their |Close()| method, which should inform us of being closed via our
   // |Close()|. Thus these should already be null.
   DCHECK(!endpoints_[0]);
   DCHECK(!endpoints_[1]);
 }
 
-MojoResult MessagePipe::EnqueueMessage(
+MojoResult MessagePipe::EnqueueMessageNoLock(
     unsigned port,
     scoped_ptr<MessageInTransit> message,
     std::vector<DispatcherTransport>* transports) {
   DCHECK(port == 0 || port == 1);
   DCHECK(message);
 
   DCHECK_EQ(message->type(), MessageInTransit::kTypeEndpoint);
-
-  base::AutoLock locker(lock_);
   DCHECK(endpoints_[GetPeerPort(port)]);
 
   // The destination port need not be open, unlike the source port.
   if (!endpoints_[port])
     return MOJO_RESULT_FAILED_PRECONDITION;
 
   if (transports) {
     MojoResult result = AttachTransportsNoLock(port, message.get(), transports);
     if (result != MOJO_RESULT_OK)
       return result;
@@ skipping 43 matching lines @@
       LOG(WARNING) << "Enqueueing null dispatcher";
       dispatchers->push_back(nullptr);
     }
   }
   message->SetDispatchers(dispatchers.Pass());
   return MOJO_RESULT_OK;
 }
 
 }  // namespace system
 }  // namespace mojo