Update mojo sdk to rev 5aa6dbdccf1950daf0cd3014bf763f35899bccf9

BUG=433814

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

Cr-Commit-Position: refs/heads/master@{#305339}

mojo/edk/js/core.cc

@@ -57,8 +57,7 @@ gin::Dictionary CreateMessagePipe(const gin::Arguments& args) {
       options_value->IsUndefined()) {
     result = MojoCreateMessagePipe(NULL, &handle0, &handle1);
   } else if (options_value->IsObject()) {
-    gin::Dictionary options_dict(args.isolate(),
-                                 options_value->ToObject(args.isolate()));
+    gin::Dictionary options_dict(args.isolate(), options_value->ToObject());
     MojoCreateMessagePipeOptions options;
     // For future struct_size, we can probably infer that from the presence of
     // properties in options_dict. For now, it's always 8.
@@ -157,8 +156,7 @@ gin::Dictionary CreateDataPipe(const gin::Arguments& args) {
       options_value->IsUndefined()) {
     result = MojoCreateDataPipe(NULL, &producer_handle, &consumer_handle);
   } else if (options_value->IsObject()) {
-    gin::Dictionary options_dict(args.isolate(),
-                                 options_value->ToObject(args.isolate()));
+    gin::Dictionary options_dict(args.isolate(), options_value->ToObject());
     MojoCreateDataPipeOptions options;
     // For future struct_size, we can probably infer that from the presence of
     // properties in options_dict. For now, it's always 16.

mojo/edk/js/waiting_callback.cc

@@ -4,6 +4,8 @@
 
 #include "mojo/edk/js/waiting_callback.h"
 
+#include "base/bind.h"
+#include "base/message_loop/message_loop.h"
 #include "gin/per_context_data.h"
 #include "mojo/public/cpp/environment/environment.h"
 
@@ -50,7 +52,7 @@ void WaitingCallback::Cancel() {
 WaitingCallback::WaitingCallback(v8::Isolate* isolate,
                                  v8::Handle<v8::Function> callback,
                                  gin::Handle<HandleWrapper> handle_wrapper)
-    : wait_id_(0), handle_wrapper_(handle_wrapper.get()) {
+    : wait_id_(0), handle_wrapper_(handle_wrapper.get()), weak_factory_(this) {
   handle_wrapper_->AddCloseObserver(this);
   v8::Handle<v8::Context> context = isolate->GetCurrentContext();
   runner_ = gin::PerContextData::From(context)->runner()->GetWeakPtr();
@@ -66,10 +68,21 @@ void WaitingCallback::CallOnHandleReady(void* closure, MojoResult result) {
   static_cast<WaitingCallback*>(closure)->OnHandleReady(result);
 }
 
-void WaitingCallback::OnHandleReady(MojoResult result) {
+void WaitingCallback::ClearWaitId() {
   wait_id_ = 0;
   handle_wrapper_->RemoveCloseObserver(this);
-  handle_wrapper_ = NULL;
+  handle_wrapper_ = nullptr;
+}
+
+void WaitingCallback::OnHandleReady(MojoResult result) {
+  ClearWaitId();
+  CallCallback(result);
+}
+
+void WaitingCallback::CallCallback(MojoResult result) {
+  // ClearWaitId must already have been called.
+  DCHECK(!wait_id_);
+  DCHECK(!handle_wrapper_);
 
   if (!runner_)
     return;
@@ -88,7 +101,14 @@ void WaitingCallback::OnHandleReady(MojoResult result) {
 
 void WaitingCallback::OnWillCloseHandle() {
   Environment::GetDefaultAsyncWaiter()->CancelWait(wait_id_);
-  OnHandleReady(MOJO_RESULT_INVALID_ARGUMENT);
+
+  // This may be called from GC, so we can't execute Javascript now, call
+  // ClearWaitId explicitly, and CallCallback asynchronously.
+  ClearWaitId();
+  base::MessageLoop::current()->PostTask(
+      FROM_HERE,
+      base::Bind(&WaitingCallback::CallCallback, weak_factory_.GetWeakPtr(),
+                 MOJO_RESULT_INVALID_ARGUMENT));
 }
 
 }  // namespace js

mojo/edk/js/waiting_callback.h

@@ -5,6 +5,7 @@
 #ifndef MOJO_EDK_JS_WAITING_CALLBACK_H_
 #define MOJO_EDK_JS_WAITING_CALLBACK_H_
 
+#include "base/memory/weak_ptr.h"
 #include "gin/handle.h"
 #include "gin/runner.h"
 #include "gin/wrappable.h"
@@ -49,10 +50,14 @@ class WaitingCallback : public gin::Wrappable<WaitingCallback>,
   // still in progress.
   void OnWillCloseHandle() override;
 
+  void ClearWaitId();
+  void CallCallback(MojoResult result);
+
   base::WeakPtr<gin::Runner> runner_;
   MojoAsyncWaitID wait_id_;
 
   HandleWrapper* handle_wrapper_;
+  base::WeakPtrFactory<WaitingCallback> weak_factory_;
 
   DISALLOW_COPY_AND_ASSIGN(WaitingCallback);
 };

mojo/edk/mojo_edk.gyp

@@ -54,6 +54,7 @@
         'system/channel.h',
         'system/channel_endpoint.cc',
         'system/channel_endpoint.h',
+        'system/channel_endpoint_client.h',
         'system/channel_endpoint_id.cc',
         'system/channel_endpoint_id.h',
         'system/channel_info.cc',

mojo/edk/mojo_edk_tests.gyp

@@ -169,6 +169,7 @@
         'embedder/platform_channel_pair_posix_unittest.cc',
         'embedder/simple_platform_shared_buffer_unittest.cc',
         'system/channel_endpoint_id_unittest.cc',
+        'system/channel_manager_unittest.cc',
         'system/channel_unittest.cc',
         'system/core_unittest.cc',
         'system/core_test_base.cc',

mojo/edk/system/BUILD.gn

@@ -17,6 +17,7 @@ component("system") {
     "channel.h",
     "channel_endpoint.cc",
     "channel_endpoint.h",
+    "channel_endpoint_client.h",
     "channel_endpoint_id.cc",
     "channel_endpoint_id.h",
     "channel_info.cc",
@@ -103,6 +104,7 @@ test("mojo_system_unittests") {
   sources = [
     "../test/multiprocess_test_helper_unittest.cc",
     "channel_endpoint_id_unittest.cc",
+    "channel_manager_unittest.cc",
     "channel_unittest.cc",
     "core_test_base.cc",
     "core_test_base.h",

mojo/edk/system/channel.cc

@@ -75,7 +75,6 @@ void Channel::Shutdown() {
        it != to_destroy.end(); ++it) {
     if (it->second.get()) {
       num_live++;
-      it->second->OnDisconnect();
       it->second->DetachFromChannel();
     } else {
       num_zombies++;
@@ -475,7 +474,6 @@ bool Channel::OnRemoveMessagePipeEndpoint(ChannelEndpointId local_id,
         static_cast<unsigned>(remote_id.value())));
   }
 
-  endpoint->OnDisconnect();
   return true;
 }
 

mojo/edk/system/channel.h

@@ -196,8 +196,15 @@ class MOJO_SYSTEM_IMPL_EXPORT Channel
 
   typedef base::hash_map<ChannelEndpointId, scoped_refptr<MessagePipe>>
       IdToMessagePipeMap;
-  // Map from local IDs to pending/incoming endpoints (i.e., those which do not
-  // yet have a dispatcher attached).
+  // Map from local IDs to pending/incoming message pipes (i.e., those which do
+  // not yet have a dispatcher attached).
+  // TODO(vtl): This is a layering violation, since |Channel| shouldn't know
+  // about |MessagePipe|. However, we can't just hang on to |ChannelEndpoint|s
+  // (even if they have a reference to the |MessagePipe|) since their lifetimes
+  // are tied to the "remote" side. When |ChannelEndpoint::DetachFromChannel()|
+  // (eventually) results in |ChannelEndpoint::DetachFromClient()| being called.
+  // We really need to hang on to the "local" side of the message pipe, to which
+  // dispatchers will be "attached".
   IdToMessagePipeMap incoming_message_pipes_;
   // TODO(vtl): We need to keep track of remote IDs (so that we don't collide
   // if/when we wrap).

mojo/edk/system/channel_endpoint.cc

@@ -6,18 +6,17 @@
 
 #include "base/logging.h"
 #include "mojo/edk/system/channel.h"
-#include "mojo/edk/system/message_pipe.h"
+#include "mojo/edk/system/channel_endpoint_client.h"
 #include "mojo/edk/system/transport_data.h"
 
 namespace mojo {
 namespace system {
 
-ChannelEndpoint::ChannelEndpoint(MessagePipe* message_pipe,
-                                 unsigned port,
+ChannelEndpoint::ChannelEndpoint(ChannelEndpointClient* client,
+                                 unsigned client_port,
                                  MessageInTransitQueue* message_queue)
-    : message_pipe_(message_pipe), port_(port), channel_(nullptr) {
-  DCHECK(message_pipe_.get() || message_queue);
-  DCHECK(port_ == 0 || port_ == 1);
+    : client_(client), client_port_(client_port), channel_(nullptr) {
+  DCHECK(client_.get() || message_queue);
 
   if (message_queue)
     paused_message_queue_.Swap(message_queue);
@@ -28,33 +27,29 @@ bool ChannelEndpoint::EnqueueMessage(scoped_ptr<MessageInTransit> message) {
 
   base::AutoLock locker(lock_);
 
-  if (!channel_ || !remote_id_.is_valid()) {
-    // We may reach here if we haven't been attached or run yet.
+  if (!channel_) {
+    // We may reach here if we haven't been attached/run yet.
     // TODO(vtl): We may also reach here if the channel is shut down early for
-    // some reason (with live message pipes on it). We can't check |state_| yet,
-    // until it's protected under lock, but in this case we should return false
-    // (and not enqueue any messages).
+    // some reason (with live message pipes on it). Ideally, we'd return false
+    // (and not enqueue the message), but we currently don't have a way to check
+    // this.
     paused_message_queue_.AddMessage(message.Pass());
     return true;
   }
 
-  // TODO(vtl): Currently, this only works in the "running" case.
-  DCHECK(remote_id_.is_valid());
-
   return WriteMessageNoLock(message.Pass());
 }
 
-void ChannelEndpoint::DetachFromMessagePipe() {
+void ChannelEndpoint::DetachFromClient() {
   {
     base::AutoLock locker(lock_);
-    DCHECK(message_pipe_.get());
-    message_pipe_ = nullptr;
+    DCHECK(client_.get());
+    client_ = nullptr;
 
     if (!channel_)
       return;
     DCHECK(local_id_.is_valid());
-    // TODO(vtl): Once we combine "run" into "attach", |remote_id_| should valid
-    // here as well.
+    DCHECK(remote_id_.is_valid());
     channel_->DetachEndpoint(this, local_id_, remote_id_);
     channel_ = nullptr;
     local_id_ = ChannelEndpointId();
@@ -82,7 +77,7 @@ void ChannelEndpoint::AttachAndRun(Channel* channel,
         << "Failed to write enqueue message to channel";
   }
 
-  if (!message_pipe_.get()) {
+  if (!client_.get()) {
     channel_->DetachEndpoint(this, local_id_, remote_id_);
     channel_ = nullptr;
     local_id_ = ChannelEndpointId();
@@ -94,12 +89,12 @@ bool ChannelEndpoint::OnReadMessage(
     const MessageInTransit::View& message_view,
     embedder::ScopedPlatformHandleVectorPtr platform_handles) {
   scoped_ptr<MessageInTransit> message(new MessageInTransit(message_view));
-  scoped_refptr<MessagePipe> message_pipe;
-  unsigned port;
+  scoped_refptr<ChannelEndpointClient> client;
+  unsigned client_port;
   {
     base::AutoLock locker(lock_);
     DCHECK(channel_);
-    if (!message_pipe_.get()) {
+    if (!client_.get()) {
       // This isn't a failure per se. (It just means that, e.g., the other end
       // of the message point closed first.)
       return true;
@@ -113,49 +108,44 @@ bool ChannelEndpoint::OnReadMessage(
           channel_));
     }
 
-    // Take a ref, and call |EnqueueMessage()| outside the lock.
-    message_pipe = message_pipe_;
-    port = port_;
+    // Take a ref, and call |OnReadMessage()| outside the lock.
+    client = client_;
+    client_port = client_port_;
   }
 
-  MojoResult result = message_pipe->EnqueueMessage(
-      MessagePipe::GetPeerPort(port), message.Pass());
-  return (result == MOJO_RESULT_OK);
+  return client->OnReadMessage(client_port, message.Pass());
 }
 
-void ChannelEndpoint::OnDisconnect() {
-  scoped_refptr<MessagePipe> message_pipe;
-  unsigned port;
+void ChannelEndpoint::DetachFromChannel() {
+  scoped_refptr<ChannelEndpointClient> client;
+  unsigned client_port = 0;
   {
     base::AutoLock locker(lock_);
-    if (!message_pipe_.get())
-      return;
 
-    // Take a ref, and call |Close()| outside the lock.
-    message_pipe = message_pipe_;
-    port = port_;
-  }
-  message_pipe->Close(port);
-}
+    if (client_.get()) {
+      // Take a ref, and call |OnDetachFromChannel()| outside the lock.
+      client = client_;
+      client_port = client_port_;
+    }
 
-void ChannelEndpoint::DetachFromChannel() {
-  base::AutoLock locker(lock_);
-  // This may already be null if we already detached from the channel in
-  // |DetachFromMessagePipe()| by calling |Channel::DetachEndpoint()| (and there
-  // are racing detaches).
-  if (!channel_)
-    return;
+    // |channel_| may already be null if we already detached from the channel in
+    // |DetachFromClient()| by calling |Channel::DetachEndpoint()| (and there
+    // are racing detaches).
+    if (channel_) {
+      DCHECK(local_id_.is_valid());
+      DCHECK(remote_id_.is_valid());
+      channel_ = nullptr;
+      local_id_ = ChannelEndpointId();
+      remote_id_ = ChannelEndpointId();
+    }
+  }
 
-  DCHECK(local_id_.is_valid());
-  // TODO(vtl): Once we combine "run" into "attach", |remote_id_| should valid
-  // here as well.
-  channel_ = nullptr;
-  local_id_ = ChannelEndpointId();
-  remote_id_ = ChannelEndpointId();
+  if (client.get())
+    client->OnDetachFromChannel(client_port);
 }
 
 ChannelEndpoint::~ChannelEndpoint() {
-  DCHECK(!message_pipe_.get());
+  DCHECK(!client_.get());
   DCHECK(!channel_);
   DCHECK(!local_id_.is_valid());
   DCHECK(!remote_id_.is_valid());