Mojo: More work towards sending message pipe handles over remote message pipes.

ProxyMessagePipeEndpoint::EnqueueMessage() needs to look at each handle
(i.e., dispatcher) and decide what to do.

This CL doesn't actually do anything yet exactly. I also need to check
(at a higher level) that we're not sending the peer handle over a given
handle, hence I'm splitting out this part now (since that work will
touch on the same things as this).

R=davemoore@chromium.org

Review URL: https://codereview.chromium.org/145483005

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@248580 0039d316-1c4b-4281-b951-d872f2087c98

mojo/system/message_pipe_dispatcher.cc

@@ -27,6 +27,16 @@ void MessagePipeDispatcher::Init(scoped_refptr<MessagePipe> message_pipe,
   port_ = port;
 }
 
+MessagePipe* MessagePipeDispatcher::GetMessagePipeNoLock() const {
+  lock().AssertAcquired();
+  return message_pipe_.get();
+}
+
+unsigned MessagePipeDispatcher::GetPortNoLock() const {
+  lock().AssertAcquired();
+  return port_;
+}
+
 Dispatcher::Type MessagePipeDispatcher::GetType() {
   return kTypeMessagePipe;
 }

mojo/system/message_pipe_dispatcher.h

@@ -25,6 +25,14 @@ class MOJO_SYSTEM_IMPL_EXPORT MessagePipeDispatcher : public Dispatcher {
   // Must be called before any other methods. (This method is not thread-safe.)
   void Init(scoped_refptr<MessagePipe> message_pipe, unsigned port);
 
+  // Gets a dumb pointer to |message_pipe_|. This must be called under the
+  // |Dispatcher| lock (that it's a dumb pointer is okay since it's under lock).
+  // This is needed when sending handles across processes, where nontrivial,
+  // invasive work needs to be done.
+  MessagePipe* GetMessagePipeNoLock() const;
+  // Similarly for the port.
+  unsigned GetPortNoLock() const;
+
   virtual Type GetType() OVERRIDE;
 
  private:

mojo/system/proxy_message_pipe_endpoint.cc

@@ -6,9 +6,11 @@
 
 #include <string.h>
 
+#include "base/containers/hash_tables.h"
 #include "base/logging.h"
 #include "base/stl_util.h"
 #include "mojo/system/channel.h"
+#include "mojo/system/message_pipe_dispatcher.h"
 
 namespace mojo {
 namespace system {
@@ -63,7 +65,15 @@ MojoResult ProxyMessagePipeEndpoint::EnqueueMessage(
     const std::vector<Dispatcher*>* dispatchers) {
   DCHECK(!dispatchers || !dispatchers->empty());
 
-  MojoResult result = CanEnqueueDispatchers(dispatchers);
+  // No need to preflight if there are no dispatchers.
+  if (!dispatchers) {
+    EnqueueMessageInternal(message, NULL);
+    return MOJO_RESULT_OK;
+  }
+
+  std::vector<PreflightDispatcherInfo> preflight_dispatcher_infos;
+  MojoResult result = PreflightDispatchers(dispatchers,
+                                           &preflight_dispatcher_infos);
   if (result != MOJO_RESULT_OK) {
     message->Destroy();
     return result;
@@ -105,14 +115,68 @@ void ProxyMessagePipeEndpoint::Run(MessageInTransit::EndpointId remote_id) {
   paused_message_queue_.clear();
 }
 
-MojoResult ProxyMessagePipeEndpoint::CanEnqueueDispatchers(
-    const std::vector<Dispatcher*>* dispatchers) {
-  // TODO(vtl): Support sending handles over OS pipes.
-  if (dispatchers) {
-    NOTIMPLEMENTED();
-    return MOJO_RESULT_UNIMPLEMENTED;
+MojoResult ProxyMessagePipeEndpoint::PreflightDispatchers(
+    const std::vector<Dispatcher*>* dispatchers,
+    std::vector<PreflightDispatcherInfo>* preflight_dispatcher_infos) {
+  DCHECK(!dispatchers || !dispatchers->empty());
+  DCHECK(preflight_dispatcher_infos);
+  DCHECK(preflight_dispatcher_infos->empty());
+
+  // Size it to fit everything.
+  preflight_dispatcher_infos->resize(dispatchers->size());
+
+  // TODO(vtl): We'll begin with limited support for sending message pipe
+  // handles. We won't support:
+  //  - sending both handles (the |hash_set| below is to detect this case and
+  //    fail gracefully);
+  //  - sending a handle whose peer is remote.
+  base::hash_set<intptr_t> message_pipes;
+
+  for (size_t i = 0; i < dispatchers->size(); i++) {
+    Dispatcher* dispatcher = (*dispatchers)[i];
+    switch (dispatcher->GetType()) {
+      case Dispatcher::kTypeUnknown:
+        LOG(ERROR) << "Unknown dispatcher type";
+        return MOJO_RESULT_INTERNAL;
+
+      case Dispatcher::kTypeMessagePipe: {
+        MessagePipeDispatcher* mp_dispatcher =
+            static_cast<MessagePipeDispatcher*>(dispatcher);
+        (*preflight_dispatcher_infos)[i].message_pipe =
+            mp_dispatcher->GetMessagePipeNoLock();
+        DCHECK((*preflight_dispatcher_infos)[i].message_pipe);
+        (*preflight_dispatcher_infos)[i].port = mp_dispatcher->GetPortNoLock();
+
+        // Check for unsupported cases (see TODO above).
+        bool is_new_element = message_pipes.insert(reinterpret_cast<intptr_t>(
+            (*preflight_dispatcher_infos)[i].message_pipe)).second;
+        if (!is_new_element) {
+          NOTIMPLEMENTED()
+              << "Sending both sides of a message pipe not yet supported";
+          return MOJO_RESULT_UNIMPLEMENTED;
+        }
+        // TODO(vtl): Check that peer isn't remote (per above TODO).
+
+        break;
+      }
+
+      case Dispatcher::kTypeDataPipeProducer:
+        NOTIMPLEMENTED() << "Sending data pipe producers not yet supported";
+        return MOJO_RESULT_UNIMPLEMENTED;
+
+      case Dispatcher::kTypeDataPipeConsumer:
+        NOTIMPLEMENTED() << "Sending data pipe consumers not yet supported";
+        return MOJO_RESULT_UNIMPLEMENTED;
+
+      default:
+        LOG(ERROR) << "Invalid or unsupported dispatcher type";
+        return MOJO_RESULT_UNIMPLEMENTED;
+    }
   }
-  return MOJO_RESULT_OK;
+
+  // TODO(vtl): Support sending handles over OS pipes.
+  NOTIMPLEMENTED();
+  return MOJO_RESULT_UNIMPLEMENTED;
 }
 
 // Note: We may have to enqueue messages even when our (local) peer isn't open

mojo/system/proxy_message_pipe_endpoint.h

@@ -21,6 +21,7 @@ namespace mojo {
 namespace system {
 
 class Channel;
+class MessagePipe;
 
 // A |ProxyMessagePipeEndpoint| connects an end of a |MessagePipe| to a
 // |Channel|, over which it transmits and receives data (to/from another
@@ -54,6 +55,14 @@ class MOJO_SYSTEM_IMPL_EXPORT ProxyMessagePipeEndpoint
   virtual void Run(MessageInTransit::EndpointId remote_id) OVERRIDE;
 
  private:
+  struct PreflightDispatcherInfo {
+    PreflightDispatcherInfo() : message_pipe(), port() {}
+
+    // For now, we only support sending message pipes, so this is simple.
+    MessagePipe* message_pipe;
+    unsigned port;
+  };
+
   bool is_attached() const {
     return !!channel_.get();
   }
@@ -62,9 +71,15 @@ class MOJO_SYSTEM_IMPL_EXPORT ProxyMessagePipeEndpoint
     return remote_id_ != MessageInTransit::kInvalidEndpointId;
   }
 
-  MojoResult CanEnqueueDispatchers(const std::vector<Dispatcher*>* dispatchers);
+  // Checks that |dispatchers| will be able to be enqueued by
+  // |EnqueueMessageInternal()| (which then MUST be called if this succeeds).
+  // |dispatchers| must be non-null and nonempty; |preflight_dispatcher_infos|
+  // must be non-null and empty.
+  MojoResult PreflightDispatchers(
+      const std::vector<Dispatcher*>* dispatchers,
+      std::vector<PreflightDispatcherInfo>* preflight_dispatcher_infos);
   // |dispatchers| should be non-null only if it's nonempty, in which case the
-  // dispatchers should have been preflighted by |CanEnqueueDispatchers()|.
+  // dispatchers should have been preflighted by |PreflightDispatchers()|.
   void EnqueueMessageInternal(MessageInTransit* message,
                               const std::vector<Dispatcher*>* dispatchers);