Add multiplexing of message pipes in the new EDK.

mojo/edk/system/message_pipe_dispatcher.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/message_pipe_dispatcher.h"
 
 #include "base/bind.h"
 #include "base/logging.h"
 #include "base/message_loop/message_loop.h"
 #include "mojo/edk/embedder/embedder_internal.h"
 #include "mojo/edk/embedder/platform_handle_utils.h"
 #include "mojo/edk/embedder/platform_shared_buffer.h"
 #include "mojo/edk/embedder/platform_support.h"
 #include "mojo/edk/system/broker.h"
 #include "mojo/edk/system/configuration.h"
 #include "mojo/edk/system/message_in_transit.h"
 #include "mojo/edk/system/options_validation.h"
 #include "mojo/edk/system/transport_data.h"
 
 namespace mojo {
 namespace edk {
 
 namespace {
 
 const size_t kInvalidMessagePipeHandleIndex = static_cast<size_t>(-1);
 
 struct MOJO_ALIGNAS(8) SerializedMessagePipeHandleDispatcher {
+  bool transferable;
+  bool write_error;
+  uint64_t pipe_id;  // If transferable is false.
+  // The following members are only set if transferable is true.
   // Could be |kInvalidMessagePipeHandleIndex| if the other endpoint of the MP
   // was closed.
   size_t platform_handle_index;
-  bool write_error;
 
   size_t shared_memory_handle_index;  // (Or |kInvalidMessagePipeHandleIndex|.)
   uint32_t shared_memory_size;
 
   size_t serialized_read_buffer_size;
   size_t serialized_write_buffer_size;
   size_t serialized_message_queue_size;
 
   // These are the FDs required as part of serializing channel_ and
   // message_queue_. This is only used on POSIX.
@@ skipping 41 matching lines @@
 
 const MojoCreateMessagePipeOptions
     MessagePipeDispatcher::kDefaultCreateOptions = {
         static_cast<uint32_t>(sizeof(MojoCreateMessagePipeOptions)),
         MOJO_CREATE_MESSAGE_PIPE_OPTIONS_FLAG_NONE};
 
 MojoResult MessagePipeDispatcher::ValidateCreateOptions(
     const MojoCreateMessagePipeOptions* in_options,
     MojoCreateMessagePipeOptions* out_options) {
   const MojoCreateMessagePipeOptionsFlags kKnownFlags =
-      MOJO_CREATE_MESSAGE_PIPE_OPTIONS_FLAG_NONE;
+      MOJO_CREATE_MESSAGE_PIPE_OPTIONS_FLAG_NONE |
+      MOJO_CREATE_MESSAGE_PIPE_OPTIONS_FLAG_TRANSFERABLE;
 
   *out_options = kDefaultCreateOptions;
   if (!in_options)
     return MOJO_RESULT_OK;
 
   UserOptionsReader<MojoCreateMessagePipeOptions> reader(in_options);
   if (!reader.is_valid())
     return MOJO_RESULT_INVALID_ARGUMENT;
 
   if (!OPTIONS_STRUCT_HAS_MEMBER(MojoCreateMessagePipeOptions, flags, reader))
     return MOJO_RESULT_OK;
   if ((reader.options().flags & ~kKnownFlags))
     return MOJO_RESULT_UNIMPLEMENTED;
   out_options->flags = reader.options().flags;
 
   // Checks for fields beyond |flags|:
 
   // (Nothing here yet.)
 
   return MOJO_RESULT_OK;
 }
 
 void MessagePipeDispatcher::Init(
     ScopedPlatformHandle message_pipe,
     char* serialized_read_buffer, size_t serialized_read_buffer_size,
     char* serialized_write_buffer, size_t serialized_write_buffer_size,
     std::vector<int>* serialized_read_fds,
     std::vector<int>* serialized_write_fds) {
+  CHECK(transferable_);
   if (message_pipe.get().is_valid()) {
     channel_ = RawChannel::Create(message_pipe.Pass());
 
     // TODO(jam): It's probably cleaner to pass this in Init call.
     channel_->SetSerializedData(
         serialized_read_buffer, serialized_read_buffer_size,
         serialized_write_buffer, serialized_write_buffer_size,
         serialized_read_fds, serialized_write_fds);
     internal::g_io_thread_task_runner->PostTask(
         FROM_HERE, base::Bind(&MessagePipeDispatcher::InitOnIO, this));
   }
 }
 
+void MessagePipeDispatcher::InitNonTransferable(uint64_t pipe_id) {
+  CHECK(!transferable_);
+  pipe_id_ = pipe_id;
+}
+
 void MessagePipeDispatcher::InitOnIO() {
   base::AutoLock locker(lock());
+  CHECK(transferable_);
   calling_init_ = true;
   if (channel_)
     channel_->Init(this);
   calling_init_ = false;
 }
 
 void MessagePipeDispatcher::CloseOnIO() {
   base::AutoLock locker(lock());
+  if (transferable_) {
+    if (channel_) {
+      channel_->Shutdown();
+      channel_ = nullptr;
+    }
+  } else {
+    if (non_transferable_state_ == CONNECT_CALLED ||
+        non_transferable_state_ == WAITING_FOR_READ_OR_WRITE) {
+      if (non_transferable_state_ == WAITING_FOR_READ_OR_WRITE)
+        RequestNontransferableChannel();
 
-  if (channel_) {
-    channel_->Shutdown();
-    channel_ = nullptr;
+      // We can't cancel the pending request yet, since the other side of the
+      // message pipe would want to get pending outgoing messages (if any) or
+      // at least know that this end was closed. So keep this object alive until
+      // then.
+      non_transferable_state_ = WAITING_FOR_CONNECT_TO_CLOSE;
+      AddRef();
+    } else if (non_transferable_state_ == CONNECTED) {
+      internal::g_broker->CloseMessagePipe(pipe_id_, this);
+      non_transferable_state_ = CLOSED;
+      channel_ = nullptr;
+    }
   }
 }
 
 Dispatcher::Type MessagePipeDispatcher::GetType() const {
   return Type::MESSAGE_PIPE;
 }
 
+void MessagePipeDispatcher::GotNonTransferableChannel(RawChannel* channel) {
+  base::AutoLock locker(lock());
+  channel_ = channel;
+  while (!non_transferable_outgoing_message_queue_.IsEmpty()) {
+    channel_->WriteMessage(
+        non_transferable_outgoing_message_queue_.GetMessage());
+  }
+
+  if (non_transferable_state_ == WAITING_FOR_CONNECT_TO_CLOSE) {
+    // We kept this object alive until it's connected, we can release it now.
+    // Since we're in a callback from the Broker, call it asynchronously.
+    internal::g_io_thread_task_runner->PostTask(
+        FROM_HERE,
+        base::Bind(&Broker::CloseMessagePipe,
+                   base::Unretained(internal::g_broker), pipe_id_,
+                   base::Unretained(this)));
+    non_transferable_state_ = CLOSED;
+    channel_ = nullptr;
+    base::MessageLoop::current()->ReleaseSoon(FROM_HERE, this);
+  } else {
+    non_transferable_state_ = CONNECTED;
+  }
+}
+
 #if defined(OS_WIN)
 // TODO(jam): this is copied from RawChannelWin till I figure out what's the
 // best way we want to share this.
 // Since this is used for serialization of messages read/written to a MP that
 // aren't consumed by Mojo primitives yet, there could be an unbounded number of
 // them when a MP is being sent. As a result, even for POSIX we will probably
 // want to send the handles to the shell process and exchange them for tokens
 // (since we can be sure that the shell will respond to our IPCs, compared to
 // the other end where we're sending the MP to, which may not be reading...).
 ScopedPlatformHandleVectorPtr GetReadPlatformHandles(
@@ skipping 13 matching lines @@
     const void* source,
     size_t size,
     PlatformHandleVector* platform_handles) {
   if (size != sizeof(SerializedMessagePipeHandleDispatcher)) {
     LOG(ERROR) << "Invalid serialized message pipe dispatcher (bad size)";
     return nullptr;
   }
 
   const SerializedMessagePipeHandleDispatcher* serialization =
       static_cast<const SerializedMessagePipeHandleDispatcher*>(source);
+
+  scoped_refptr<MessagePipeDispatcher> rv(
+      new MessagePipeDispatcher(serialization->transferable));
+  if (!rv->transferable_) {
+    rv->InitNonTransferable(serialization->pipe_id);
+    return rv;
+  }
+
   if (serialization->shared_memory_size !=
       (serialization->serialized_read_buffer_size +
        serialization->serialized_write_buffer_size +
        serialization->serialized_message_queue_size)) {
     LOG(ERROR) << "Invalid serialized message pipe dispatcher (bad struct)";
     return nullptr;
   }
 
   ScopedPlatformHandle platform_handle, shared_memory_handle;
   if (!GetHandle(serialization->platform_handle_index,
@@ skipping 27 matching lines @@
           serialization->serialized_write_buffer_size;
       buffer += serialized_write_buffer_size;
     }
     if (serialization->serialized_message_queue_size) {
       message_queue_data = buffer;
       message_queue_size = serialization->serialized_message_queue_size;
       buffer += message_queue_size;
     }
   }
 
-  scoped_refptr<MessagePipeDispatcher> rv(
-      Create(MessagePipeDispatcher::kDefaultCreateOptions));
   rv->write_error_ = serialization->write_error;
 
   std::vector<int> serialized_read_fds;
   std::vector<int> serialized_write_fds;
 #if defined(OS_POSIX)
   std::vector<int> serialized_fds;
   size_t serialized_fds_index = 0;
 
   size_t total_fd_count = serialization->serialized_read_fds_length +
                           serialization->serialized_write_fds_length +
@@ skipping 14 matching lines @@
   serialized_fds_index += serialization->serialized_read_fds_length;
   serialized_write_fds.assign(
       serialized_fds.begin() + serialized_fds_index,
       serialized_fds.begin() + serialized_fds_index +
           serialization->serialized_write_fds_length);
   serialized_fds_index += serialization->serialized_write_fds_length;
 #endif
 
   while (message_queue_size) {
     size_t message_size;
-    CHECK(MessageInTransit::GetNextMessageSize(
-              message_queue_data, message_queue_size, &message_size));
+    if (!MessageInTransit::GetNextMessageSize(
+            message_queue_data, message_queue_size, &message_size)) {
+      NOTREACHED() << "Couldn't read message size from serialized data.";
+      return nullptr;
+    }
+    if (message_size > message_queue_size) {
+      NOTREACHED() << "Invalid serialized message size.";
+      return nullptr;
+    }
     MessageInTransit::View message_view(message_size, message_queue_data);
     message_queue_size -= message_size;
     message_queue_data += message_size;
 
     // TODO(jam): Copied below from RawChannelWin. See commment above
     // GetReadPlatformHandles.
     ScopedPlatformHandleVectorPtr temp_platform_handles;
     if (message_view.transport_data_buffer()) {
       size_t num_platform_handles;
       const void* platform_handle_table;
@@ skipping 42 matching lines @@
            &serialized_write_fds);
 
   if (message_queue_size) {  // Should be empty by now.
     LOG(ERROR) << "Invalid queued messages";
     return nullptr;
   }
 
   return rv;
 }
 
-MessagePipeDispatcher::MessagePipeDispatcher()
+MessagePipeDispatcher::MessagePipeDispatcher(bool transferable)
     : channel_(nullptr),
-      serialized_(false),
       serialized_read_fds_length_(0u),
       serialized_write_fds_length_(0u),
       serialized_message_fds_length_(0u),
+      pipe_id_(0),
+      non_transferable_state_(WAITING_FOR_READ_OR_WRITE),
+      serialized_(false),
       calling_init_(false),
-      write_error_(false) {
+      write_error_(false),
+      transferable_(transferable) {
 }
 
 MessagePipeDispatcher::~MessagePipeDispatcher() {
   // |Close()|/|CloseImplNoLock()| should have taken care of the channel. The
   // exception is if they posted a task to run CloseOnIO but the IO thread shut
   // down and so when it was deleting pending tasks it caused the last reference
   // to destruct this object. In that case, safe to destroy the channel.
   if (channel_ && internal::g_io_thread_task_runner->RunsTasksOnCurrentThread())
     channel_->Shutdown();
   else
     DCHECK(!channel_);
 #if defined(OS_POSIX)
   ClosePlatformHandles(&serialized_fds_);
 #endif
 }
 
 void MessagePipeDispatcher::CancelAllAwakablesNoLock() {
   lock().AssertAcquired();
   awakable_list_.CancelAll();
 }
 
 void MessagePipeDispatcher::CloseImplNoLock() {
   lock().AssertAcquired();
   internal::g_io_thread_task_runner->PostTask(
       FROM_HERE, base::Bind(&MessagePipeDispatcher::CloseOnIO, this));
 }
 
 void MessagePipeDispatcher::SerializeInternal() {
+  serialized_ = true;
+  if (!transferable_) {
+    CHECK(non_transferable_state_ == WAITING_FOR_READ_OR_WRITE)
+        << "Non transferable message pipe being sent after read/write. "
+        << "MOJO_CREATE_MESSAGE_PIPE_OPTIONS_FLAG_TRANSFERABLE must be used if "
+        << "the pipe can be sent after it's read or written.";
+    non_transferable_state_ = SERIALISED;
+    return;
+  }
+
   // We need to stop watching handle immediately, even though not on IO thread,
   // so that other messages aren't read after this.
   std::vector<int> serialized_read_fds, serialized_write_fds;
   if (channel_) {
     bool write_error = false;
 
     serialized_platform_handle_ = channel_->ReleaseHandle(
         &serialized_read_buffer_, &serialized_write_buffer_,
         &serialized_read_fds, &serialized_write_fds, &write_error);
     serialized_fds_.insert(serialized_fds_.end(), serialized_read_fds.begin(),
                           serialized_read_fds.end());
     serialized_read_fds_length_ = serialized_read_fds.size();
     serialized_fds_.insert(serialized_fds_.end(), serialized_write_fds.begin(),
                           serialized_write_fds.end());
     serialized_write_fds_length_ = serialized_write_fds.size();
     channel_ = nullptr;
-    if (write_error)
-      write_error = true;
   } else {
     // It's valid that the other side wrote some data and closed its end.
   }
 
   DCHECK(serialized_message_queue_.empty());
   while (!message_queue_.IsEmpty()) {
     scoped_ptr<MessageInTransit> message = message_queue_.GetMessage();
 
     // When MojoWriteMessage is called, the MessageInTransit doesn't have
     // dispatchers set and CreateEquivaent... is called since the dispatchers
@@ skipping 34 matching lines @@
             &all_platform_handles->at(0), all_platform_handles->size(), tokens);
         for (size_t i = 0; i < all_platform_handles->size(); i++)
           all_platform_handles->at(i) = PlatformHandle();
 #else
         for (size_t i = 0; i < all_platform_handles->size(); i++) {
           serialized_fds_.push_back(all_platform_handles->at(i).fd);
           serialized_message_fds_length_++;
           all_platform_handles->at(i) = PlatformHandle();
         }
 #endif
+      }
 
       serialized_message_queue_.insert(
           serialized_message_queue_.end(),
           static_cast<const char*>(message->transport_data()->buffer()),
           static_cast<const char*>(message->transport_data()->buffer()) +
               transport_data_buffer_size);
-      }
     }
 
     for (size_t i = 0; i < dispatchers.size(); ++i)
       dispatchers[i]->TransportEnded();
   }
-
-  serialized_ = true;
 }
 
 scoped_refptr<Dispatcher>
 MessagePipeDispatcher::CreateEquivalentDispatcherAndCloseImplNoLock() {
   lock().AssertAcquired();
 
   SerializeInternal();
 
-  // TODO(vtl): Currently, there are no options, so we just use
-  // |kDefaultCreateOptions|. Eventually, we'll have to duplicate the options
-  // too.
-  scoped_refptr<MessagePipeDispatcher> rv = Create(kDefaultCreateOptions);
-  rv->serialized_platform_handle_ = serialized_platform_handle_.Pass();
-  serialized_message_queue_.swap(rv->serialized_message_queue_);
-  serialized_read_buffer_.swap(rv->serialized_read_buffer_);
-  serialized_write_buffer_.swap(rv->serialized_write_buffer_);
-  serialized_fds_.swap(rv->serialized_fds_);
-  rv->serialized_read_fds_length_ = serialized_read_fds_length_;
-  rv->serialized_write_fds_length_ = serialized_write_fds_length_;
-  rv->serialized_message_fds_length_ = serialized_message_fds_length_;
+  scoped_refptr<MessagePipeDispatcher> rv(
+      new MessagePipeDispatcher(transferable_));
   rv->serialized_ = true;
-  rv->write_error_ = write_error_;
-  return scoped_refptr<Dispatcher>(rv.get());
+  if (transferable_) {
+    rv->serialized_platform_handle_ = serialized_platform_handle_.Pass();
+    serialized_message_queue_.swap(rv->serialized_message_queue_);
+    serialized_read_buffer_.swap(rv->serialized_read_buffer_);
+    serialized_write_buffer_.swap(rv->serialized_write_buffer_);
+    serialized_fds_.swap(rv->serialized_fds_);
+    rv->serialized_read_fds_length_ = serialized_read_fds_length_;
+    rv->serialized_write_fds_length_ = serialized_write_fds_length_;
+    rv->serialized_message_fds_length_ = serialized_message_fds_length_;
+    rv->write_error_ = write_error_;
+  } else {
+    rv->pipe_id_ = pipe_id_;
+    rv->non_transferable_state_ = non_transferable_state_;
+  }
+  return rv;
 }
 
 MojoResult MessagePipeDispatcher::WriteMessageImplNoLock(
     const void* bytes,
     uint32_t num_bytes,
     std::vector<DispatcherTransport>* transports,
     MojoWriteMessageFlags flags) {
+  lock().AssertAcquired();
 
   DCHECK(!transports ||
          (transports->size() > 0 &&
           transports->size() <= GetConfiguration().max_message_num_handles));
 
-  lock().AssertAcquired();
-
-  if (!channel_ || write_error_)
+  if (write_error_ ||
+      (transferable_ && !channel_) ||
+      (!transferable_ && non_transferable_state_ == CLOSED)) {
     return MOJO_RESULT_FAILED_PRECONDITION;
+  }
 
   if (num_bytes > GetConfiguration().max_message_num_bytes)
     return MOJO_RESULT_RESOURCE_EXHAUSTED;
   scoped_ptr<MessageInTransit> message(new MessageInTransit(
       MessageInTransit::Type::MESSAGE, num_bytes, bytes));
   if (transports) {
     MojoResult result = AttachTransportsNoLock(message.get(), transports);
     if (result != MOJO_RESULT_OK)
       return result;
   }
 
   message->SerializeAndCloseDispatchers();
-  channel_->WriteMessage(message.Pass());
+  if (!transferable_)
+    message->set_route_id(pipe_id_);
+  if (!transferable_ &&
+      (non_transferable_state_ == WAITING_FOR_READ_OR_WRITE ||
+       non_transferable_state_ == CONNECT_CALLED)) {
+    if (non_transferable_state_ == WAITING_FOR_READ_OR_WRITE)
+      RequestNontransferableChannel();
+    non_transferable_outgoing_message_queue_.AddMessage(message.Pass());
+  } else {
+    channel_->WriteMessage(message.Pass());
+  }
 
   return MOJO_RESULT_OK;
 }
 
 MojoResult MessagePipeDispatcher::ReadMessageImplNoLock(
     void* bytes,
     uint32_t* num_bytes,
     DispatcherVector* dispatchers,
     uint32_t* num_dispatchers,
     MojoReadMessageFlags flags) {
   lock().AssertAcquired();
-  if (channel_)
+  if (channel_) {
     channel_->EnsureLazyInitialized();
+  } else if (!transferable_) {
+      if (non_transferable_state_ == WAITING_FOR_READ_OR_WRITE) {
+        RequestNontransferableChannel();
+        return MOJO_RESULT_SHOULD_WAIT;
+      } else if (non_transferable_state_ == CONNECT_CALLED) {
+        return MOJO_RESULT_SHOULD_WAIT;
+      }
+  }
+
   DCHECK(!dispatchers || dispatchers->empty());
 
   const uint32_t max_bytes = !num_bytes ? 0 : *num_bytes;
   const uint32_t max_num_dispatchers = num_dispatchers ? *num_dispatchers : 0;
 
   if (message_queue_.IsEmpty())
     return channel_ ? MOJO_RESULT_SHOULD_WAIT : MOJO_RESULT_FAILED_PRECONDITION;
 
   // TODO(vtl): If |flags & MOJO_READ_MESSAGE_FLAG_MAY_DISCARD|, we could pop
   // and release the lock immediately.
@@ skipping 43 matching lines @@
   return MOJO_RESULT_OK;
 }
 
 HandleSignalsState MessagePipeDispatcher::GetHandleSignalsStateImplNoLock()
     const {
   lock().AssertAcquired();
 
   HandleSignalsState rv;
   if (!message_queue_.IsEmpty())
     rv.satisfied_signals |= MOJO_HANDLE_SIGNAL_READABLE;
-  if (channel_ || !message_queue_.IsEmpty())
+  if (!message_queue_.IsEmpty() ||
+      (transferable_ && channel_) ||
+      (!transferable_ && non_transferable_state_ != CLOSED))
     rv.satisfiable_signals |= MOJO_HANDLE_SIGNAL_READABLE;
-  if (channel_ && !write_error_) {
+  if (!write_error_ &&
+      ((transferable_ && channel_) ||
+       (!transferable_ && non_transferable_state_ != CLOSED))) {
     rv.satisfied_signals |= MOJO_HANDLE_SIGNAL_WRITABLE;
     rv.satisfiable_signals |= MOJO_HANDLE_SIGNAL_WRITABLE;
   }
-  if (!channel_ || write_error_)
+  if (write_error_ ||
+      (transferable_ && !channel_) ||
+       (!transferable_ &&
+        ((non_transferable_state_ == CLOSED) || is_closed()))) {
     rv.satisfied_signals |= MOJO_HANDLE_SIGNAL_PEER_CLOSED;
+  }
   rv.satisfiable_signals |= MOJO_HANDLE_SIGNAL_PEER_CLOSED;
   return rv;
 }
 
 MojoResult MessagePipeDispatcher::AddAwakableImplNoLock(
     Awakable* awakable,
     MojoHandleSignals signals,
     uintptr_t context,
     HandleSignalsState* signals_state) {
   lock().AssertAcquired();
-  if (channel_)
+  if (channel_) {
     channel_->EnsureLazyInitialized();
+  } else if (!transferable_ &&
+             non_transferable_state_ == WAITING_FOR_READ_OR_WRITE) {
+    RequestNontransferableChannel();
+  }
+
   HandleSignalsState state = GetHandleSignalsStateImplNoLock();
   if (state.satisfies(signals)) {
     if (signals_state)
       *signals_state = state;
     return MOJO_RESULT_ALREADY_EXISTS;
   }
   if (!state.can_satisfy(signals)) {
     if (signals_state)
       *signals_state = state;
     return MOJO_RESULT_FAILED_PRECONDITION;
@@ skipping 30 matching lines @@
   *max_size = sizeof(SerializedMessagePipeHandleDispatcher);
 }
 
 bool MessagePipeDispatcher::EndSerializeAndCloseImplNoLock(
     void* destination,
     size_t* actual_size,
     PlatformHandleVector* platform_handles) {
   CloseImplNoLock();
   SerializedMessagePipeHandleDispatcher* serialization =
       static_cast<SerializedMessagePipeHandleDispatcher*>(destination);
+  serialization->transferable = transferable_;
+  serialization->pipe_id = pipe_id_;
   if (serialized_platform_handle_.is_valid()) {
     serialization->platform_handle_index = platform_handles->size();
     platform_handles->push_back(serialized_platform_handle_.release());
   } else {
     serialization->platform_handle_index = kInvalidMessagePipeHandleIndex;
   }
 
   serialization->write_error = write_error_;
   serialization->serialized_read_buffer_size = serialized_read_buffer_.size();
   serialization->serialized_write_buffer_size = serialized_write_buffer_.size();
@@ skipping 123 matching lines @@
       break;
   }
 
   if (started_transport_.Try()) {
     base::AutoLock locker(lock());
     // We can get two OnError callbacks before the post task below completes.
     // Although RawChannel still has a pointer to this object until Shutdown is
     // called, that is safe since this class always does a PostTask to the IO
     // thread to self destruct.
     if (channel_ && error != ERROR_WRITE) {
-      channel_->Shutdown();
+      if (transferable_) {
+        channel_->Shutdown();
+      } else {
+        CHECK_NE(non_transferable_state_, CLOSED);
+        // Since we're in a callback from the Broker, call it asynchronously.
+        internal::g_io_thread_task_runner->PostTask(
+            FROM_HERE,
+            base::Bind(&Broker::CloseMessagePipe,
+                       base::Unretained(internal::g_broker), pipe_id_,
+                       base::Unretained(this)));
+        non_transferable_state_ = CLOSED;
+      }
       channel_ = nullptr;
     }
     awakable_list_.AwakeForStateChange(GetHandleSignalsStateImplNoLock());
     started_transport_.Release();
   } else {
     // We must be waiting to call ReleaseHandle. It will call Shutdown.
   }
 }
 
 MojoResult MessagePipeDispatcher::AttachTransportsNoLock(
     MessageInTransit* message,
     std::vector<DispatcherTransport>* transports) {
   DCHECK(!message->has_dispatchers());
 
   // You're not allowed to send either handle to a message pipe over the message
   // pipe, so check for this. (The case of trying to write a handle to itself is
   // taken care of by |Core|. That case kind of makes sense, but leads to
   // complications if, e.g., both sides try to do the same thing with their
   // respective handles simultaneously. The other case, of trying to write the
   // peer handle to a handle, doesn't make sense -- since no handle will be
   // available to read the message from.)
   for (size_t i = 0; i < transports->size(); i++) {
     if (!(*transports)[i].is_valid())
       continue;
     if ((*transports)[i].GetType() == Dispatcher::Type::MESSAGE_PIPE) {
       MessagePipeDispatcher* mp =
           static_cast<MessagePipeDispatcher*>(((*transports)[i]).dispatcher());
-      if (channel_ && mp->channel_ && channel_->IsOtherEndOf(mp->channel_)) {
+      if (transferable_ && mp->transferable_ &&
+          channel_ && mp->channel_ && channel_->IsOtherEndOf(mp->channel_)) {
         // The other case should have been disallowed by |Core|. (Note: |port|
         // is the peer port of the handle given to |WriteMessage()|.)
         return MOJO_RESULT_INVALID_ARGUMENT;
+      } else if (!transferable_ && !mp->transferable_ &&
+                 pipe_id_ == mp->pipe_id_) {
+        return MOJO_RESULT_INVALID_ARGUMENT;
       }
     }
   }
 
   // Clone the dispatchers and attach them to the message. (This must be done as
   // a separate loop, since we want to leave the dispatchers alone on failure.)
   scoped_ptr<DispatcherVector> dispatchers(new DispatcherVector());
   dispatchers->reserve(transports->size());
   for (size_t i = 0; i < transports->size(); i++) {
     if ((*transports)[i].is_valid()) {
       dispatchers->push_back(
           (*transports)[i].CreateEquivalentDispatcherAndClose());
     } else {
       LOG(WARNING) << "Enqueueing null dispatcher";
       dispatchers->push_back(nullptr);
     }
   }
   message->SetDispatchers(dispatchers.Pass());
   return MOJO_RESULT_OK;
 }
 
+void MessagePipeDispatcher::RequestNontransferableChannel() {
+  lock().AssertAcquired();
+  CHECK(!transferable_);
+  CHECK_EQ(non_transferable_state_, WAITING_FOR_READ_OR_WRITE);
+  non_transferable_state_ = CONNECT_CALLED;
+
+  // PostTask since the broker can call us back synchronously.
+  internal::g_io_thread_task_runner->PostTask(
+      FROM_HERE,
+      base::Bind(&Broker::ConnectMessagePipe,
+                 base::Unretained(internal::g_broker), pipe_id_,
+                 base::Unretained(this)));
+}
+
 }  // namespace edk
 }  // namespace mojo