Mojo: Change the way message pipes are passed over channels.

mojo/edk/system/channel.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/channel.h"
 
 #include <algorithm>
 
 #include "base/bind.h"
 #include "base/compiler_specific.h"
 #include "base/logging.h"
 #include "base/macros.h"
 #include "base/strings/stringprintf.h"
 #include "mojo/edk/embedder/platform_handle_vector.h"
-#include "mojo/edk/system/message_pipe_endpoint.h"
 #include "mojo/edk/system/transport_data.h"
 
 namespace mojo {
 namespace system {
 
 Channel::Channel(embedder::PlatformSupport* platform_support)
     : platform_support_(platform_support),
       is_running_(false),
       is_shutting_down_(false) {
 }
@@ skipping 92 matching lines @@
 
     DLOG_IF(WARNING, is_shutting_down_)
         << "RunMessagePipeEndpoint() while shutting down";
   }
 
   // TODO(vtl): FIXME -- We need to handle the case that message pipe is already
   // running when we're here due to |kSubtypeChannelRunMessagePipeEndpoint|).
   endpoint->Run(remote_id);
 }
 
-void Channel::AttachAndRunEndpoint(scoped_refptr<ChannelEndpoint> endpoint,
-                                   bool is_bootstrap) {
+ChannelEndpointId Channel::AttachAndRunEndpoint(
+    scoped_refptr<ChannelEndpoint> endpoint,
+    bool is_bootstrap) {
   DCHECK(endpoint.get());
 
   ChannelEndpointId local_id;
   ChannelEndpointId remote_id;
   {
     base::AutoLock locker(lock_);
 
     DLOG_IF(WARNING, is_shutting_down_)
         << "AttachEndpoint() while shutting down";
 
     if (is_bootstrap) {
       local_id = ChannelEndpointId::GetBootstrap();
       DCHECK(local_id_to_endpoint_map_.find(local_id) ==
              local_id_to_endpoint_map_.end());
 
       remote_id = ChannelEndpointId::GetBootstrap();
     } else {
-      // TODO(vtl): More work needs to be done to enable the non-bootstrap case.
-      NOTREACHED() << "Non-bootstrap case not yet fully implemented";
       do {
         local_id = local_id_generator_.GetNext();
       } while (local_id_to_endpoint_map_.find(local_id) !=
                local_id_to_endpoint_map_.end());
 
       // TODO(vtl): We also need to check for collisions of remote IDs here.
       remote_id = remote_id_generator_.GetNext();
     }
 
     local_id_to_endpoint_map_[local_id] = endpoint;
   }
 
+  if (!is_bootstrap) {
+    if (!SendControlMessage(
+            MessageInTransit::kSubtypeChannelAttachAndRunEndpoint,
+            local_id,
+            remote_id)) {
+      HandleLocalError(base::StringPrintf(
+          "Failed to send message to run remote message pipe endpoint (local "
+          "ID %u, remote ID %u)",
+          static_cast<unsigned>(local_id.value()),
+          static_cast<unsigned>(remote_id.value())));
+      // TODO(vtl): Should we continue on to |AttachAndRun()|?
+    }
+  }
+
   endpoint->AttachAndRun(this, local_id, remote_id);
-}
-
-void Channel::RunRemoteMessagePipeEndpoint(ChannelEndpointId local_id,
-                                           ChannelEndpointId remote_id) {
-#if DCHECK_IS_ON
-  {
-    base::AutoLock locker(lock_);
-    DCHECK(local_id_to_endpoint_map_.find(local_id) !=
-           local_id_to_endpoint_map_.end());
-  }
-#endif
-
-  if (!SendControlMessage(
-          MessageInTransit::kSubtypeChannelRunMessagePipeEndpoint,
-          local_id,
-          remote_id)) {
-    HandleLocalError(base::StringPrintf(
-        "Failed to send message to run remote message pipe endpoint (local ID "
-        "%u, remote ID %u)",
-        static_cast<unsigned>(local_id.value()),
-        static_cast<unsigned>(remote_id.value())));
-  }
+  return remote_id;
 }
 
 bool Channel::WriteMessage(scoped_ptr<MessageInTransit> message) {
   base::AutoLock locker(lock_);
   if (!is_running_) {
     // TODO(vtl): I think this is probably not an error condition, but I should
     // think about it (and the shutdown sequence) more carefully.
     LOG(WARNING) << "WriteMessage() after shutdown";
     return false;
   }
@@ skipping 42 matching lines @@
           local_id,
           remote_id)) {
     HandleLocalError(base::StringPrintf(
         "Failed to send message to remove remote message pipe endpoint (local "
         "ID %u, remote ID %u)",
         static_cast<unsigned>(local_id.value()),
         static_cast<unsigned>(remote_id.value())));
   }
 }
 
+scoped_refptr<MessagePipe> Channel::PassIncomingMessagePipe(
+    ChannelEndpointId local_id) {
+  // No need to check the validity of |local_id| -- if it's not valid, it simply
+  // won't be in |incoming_message_pipes_|.
+  DVLOG_IF(2, !local_id.is_valid() || !local_id.is_remote())
+      << "Attempt to get invalid incoming message pipe for ID " << local_id;
+
+  base::AutoLock locker(lock_);
+
+  auto it = incoming_message_pipes_.find(local_id);
+  if (it == incoming_message_pipes_.end())
+    return scoped_refptr<MessagePipe>();
+
+  scoped_refptr<MessagePipe> rv;
+  rv.swap(it->second);
+  incoming_message_pipes_.erase(it);
+  return rv;
+}
+
 size_t Channel::GetSerializedPlatformHandleSize() const {
   return raw_channel_->GetSerializedPlatformHandleSize();
 }
 
 Channel::~Channel() {
   // The channel should have been shut down first.
   DCHECK(!is_running_);
 }
 
 void Channel::OnReadMessage(
@@ skipping 112 matching lines @@
 
   // Currently, no channel messages take platform handles.
   if (platform_handles) {
     HandleRemoteError(
         "Received invalid channel message (has platform handles)");
     NOTREACHED();
     return;
   }
 
   switch (message_view.subtype()) {
-    case MessageInTransit::kSubtypeChannelRunMessagePipeEndpoint:
-      DVLOG(2) << "Handling channel message to run message pipe (local ID "
-               << message_view.destination_id() << ", remote ID "
-               << message_view.source_id() << ")";
-      if (!OnRunMessagePipeEndpoint(message_view.destination_id(),
-                                    message_view.source_id())) {
+    case MessageInTransit::kSubtypeChannelAttachAndRunEndpoint:
+      DVLOG(2) << "Handling channel message to attach and run message pipe "
+                  "(local ID " << message_view.destination_id()
+               << ", remote ID " << message_view.source_id() << ")";
+      if (!OnAttachAndRunEndpoint(message_view.destination_id(),
+                                  message_view.source_id())) {
         HandleRemoteError(
-            "Received invalid channel message to run message pipe");
+            "Received invalid channel message to attach and run message pipe");
       }
       break;
     case MessageInTransit::kSubtypeChannelRemoveMessagePipeEndpoint:
       DVLOG(2) << "Handling channel message to remove message pipe (local ID "
                << message_view.destination_id() << ", remote ID "
                << message_view.source_id() << ")";
       if (!OnRemoveMessagePipeEndpoint(message_view.destination_id(),
                                        message_view.source_id())) {
         HandleRemoteError(
             "Received invalid channel message to remove message pipe");
       }
       break;
     case MessageInTransit::kSubtypeChannelRemoveMessagePipeEndpointAck:
       DVLOG(2) << "Handling channel message to ack remove message pipe (local "
                   "ID " << message_view.destination_id() << ", remote ID "
                << message_view.source_id() << ")";
       if (!OnRemoveMessagePipeEndpointAck(message_view.destination_id())) {
         HandleRemoteError(
             "Received invalid channel message to ack remove message pipe");
       }
       break;
     default:
       HandleRemoteError("Received invalid channel message");
       NOTREACHED();
       break;
   }
 }
 
-bool Channel::OnRunMessagePipeEndpoint(ChannelEndpointId local_id,
-                                       ChannelEndpointId remote_id) {
+bool Channel::OnAttachAndRunEndpoint(ChannelEndpointId local_id,
+                                     ChannelEndpointId remote_id) {
+  // We should only get this for remotely-created local endpoints, so our local
+  // ID should be "remote".
+  if (!local_id.is_valid() || !local_id.is_remote()) {
+    DVLOG(2) << "Received attach and run endpoint with invalid local ID";
+    return false;
+  }
+
+  // Conversely, the remote end should be "local".
+  if (!remote_id.is_valid() || remote_id.is_remote()) {
+    DVLOG(2) << "Received attach and run endpoint with invalid remote ID";
+    return false;
+  }
+
+  // Create a message pipe and thus an endpoint (outside the lock).
   scoped_refptr<ChannelEndpoint> endpoint;
+  scoped_refptr<MessagePipe> message_pipe(
+      MessagePipe::CreateLocalProxy(&endpoint));
+
+  bool success = true;
   {
     base::AutoLock locker(lock_);
 
-    IdToEndpointMap::iterator it = local_id_to_endpoint_map_.find(local_id);
-    if (it == local_id_to_endpoint_map_.end())
-      return false;
+    if (local_id_to_endpoint_map_.find(local_id) ==
+        local_id_to_endpoint_map_.end()) {
+      DCHECK(incoming_message_pipes_.find(local_id) ==
+             incoming_message_pipes_.end());
 
-    endpoint = it->second;
+      // TODO(vtl): Use emplace when we move to C++11 unordered_maps. (It'll
+      // avoid some refcount churn.)
+      local_id_to_endpoint_map_[local_id] = endpoint;
+      incoming_message_pipes_[local_id] = message_pipe;
+    } else {
+      // We need to call |Close()| on the message pipe outside the lock.
+      success = false;
+    }
+  }
+  if (!success) {
+    DVLOG(2) << "Received attach and run endpoint for existing local ID";
+    message_pipe->Close(0);
+    return false;
   }
 
-  RunEndpoint(endpoint, remote_id);
+  endpoint->AttachAndRun(this, local_id, remote_id);
   return true;
 }
 
 bool Channel::OnRemoveMessagePipeEndpoint(ChannelEndpointId local_id,
                                           ChannelEndpointId remote_id) {
   DCHECK(creation_thread_checker_.CalledOnValidThread());
 
   scoped_refptr<ChannelEndpoint> endpoint;
   {
     base::AutoLock locker(lock_);
@@ skipping 73 matching lines @@
   // TODO(vtl): Is this how we really want to handle this?
   // Sometimes we'll want to propagate the error back to the message pipe
   // (endpoint), and notify it that the remote is (effectively) closed.
   // Sometimes we'll want to kill the channel (and notify all the endpoints that
   // their remotes are dead.
   LOG(WARNING) << error_message;
 }
 
 }  // namespace system
 }  // namespace mojo