Mojo: Make some attempts towards fixing remote message pipe closure.

mojo/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/system/message_pipe.h"
 
 #include "base/logging.h"
 #include "mojo/system/channel.h"
 #include "mojo/system/local_message_pipe_endpoint.h"
 #include "mojo/system/message_in_transit.h"
@@ skipping 39 matching lines @@
 
 void MessagePipe::Close(unsigned port) {
   DCHECK(port == 0 || port == 1);
 
   unsigned destination_port = GetPeerPort(port);
 
   base::AutoLock locker(lock_);
   DCHECK(endpoints_[port].get());
 
   endpoints_[port]->Close();
-  if (endpoints_[destination_port].get())
-    endpoints_[destination_port]->OnPeerClose();
+  if (endpoints_[destination_port].get()) {
+    if (!endpoints_[destination_port]->OnPeerClose())
+      endpoints_[destination_port].reset();
+  }
   endpoints_[port].reset();
 }
 
 // TODO(vtl): Handle flags.
 MojoResult MessagePipe::WriteMessage(
     unsigned port,
     const void* bytes,
     uint32_t num_bytes,
     std::vector<DispatcherTransport>* transports,
     MojoWriteMessageFlags flags) {
@@ skipping 144 matching lines @@
       }
     }
     message->SetDispatchers(dispatchers.Pass());
   }
 
   // The endpoint's |EnqueueMessage()| may not report failure.
   endpoints_[port]->EnqueueMessage(message.Pass());
   return MOJO_RESULT_OK;
 }
 
-void MessagePipe::Attach(unsigned port,
+bool MessagePipe::Attach(unsigned port,
                          scoped_refptr<Channel> channel,
                          MessageInTransit::EndpointId local_id) {
   DCHECK(port == 0 || port == 1);
   DCHECK(channel.get());
   DCHECK_NE(local_id, MessageInTransit::kInvalidEndpointId);
 
   base::AutoLock locker(lock_);
-  DCHECK(endpoints_[port].get());
+  if (!endpoints_[port].get())
+    return false;
 
+  DCHECK_EQ(endpoints_[port]->GetType(), MessagePipeEndpoint::kTypeProxy);
   endpoints_[port]->Attach(channel, local_id);
+  return true;
 }
 
 void MessagePipe::Run(unsigned port, MessageInTransit::EndpointId remote_id) {
   DCHECK(port == 0 || port == 1);
   DCHECK_NE(remote_id, MessageInTransit::kInvalidEndpointId);
 
   base::AutoLock locker(lock_);
   DCHECK(endpoints_[port].get());
-  endpoints_[port]->Run(remote_id);
+  if (!endpoints_[port]->Run(remote_id))
+    endpoints_[port].reset();
+}
+
+void MessagePipe::OnRemove(unsigned port) {
+  unsigned destination_port = GetPeerPort(port);
+
+  base::AutoLock locker(lock_);
+  // A |OnPeerClose()| can come in first, before |OnRemove()| gets called.
+  if (!endpoints_[port].get())
+    return;
+
+  endpoints_[port]->OnRemove();
+  if (endpoints_[destination_port].get()) {
+    if (!endpoints_[destination_port]->OnPeerClose())
+      endpoints_[destination_port].reset();
+  }
+  endpoints_[port].reset();
 }
 
 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].get());
   DCHECK(!endpoints_[1].get());
 }
 
 MojoResult MessagePipe::HandleControlMessage(
-    unsigned port,
+    unsigned /*port*/,
     scoped_ptr<MessageInTransit> message) {
-  DCHECK(port == 0 || port == 1);
-  DCHECK(message.get());
-  DCHECK_EQ(message->type(), MessageInTransit::kTypeMessagePipe);
-
-  switch (message->subtype()) {
-    case MessageInTransit::kSubtypeMessagePipePeerClosed: {
-      unsigned source_port = GetPeerPort(port);
-
-      base::AutoLock locker(lock_);
-      DCHECK(endpoints_[source_port].get());
-
-      endpoints_[source_port]->Close();
-      if (endpoints_[port].get())
-        endpoints_[port]->OnPeerClose();
-
-      endpoints_[source_port].reset();
-      return MOJO_RESULT_OK;
-    }
-  }
-
   LOG(WARNING) << "Unrecognized MessagePipe control message subtype "
                << message->subtype();
   return MOJO_RESULT_UNKNOWN;
 }
 
 }  // namespace system
 }  // namespace mojo