Adds Channel-associated interface support on ChannelProxy's thread

ipc/ipc_mojo_bootstrap.cc

 // Copyright 2014 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 "ipc/ipc_mojo_bootstrap.h"
 
 #include <stdint.h>
 
 #include <map>
 #include <memory>
@@ skipping 424 matching lines @@
 
     mojo::InterfaceEndpointClient* client = endpoint->client();
     if (!client || !endpoint->task_runner()->BelongsToCurrentThread()) {
       // No client has been bound yet or the client runs tasks on another
       // thread. We assume the other thread must always be the one on which
       // |proxy_task_runner_| runs tasks, since that's the only valid scenario.
       //
       // If the client is not yet bound, it must be bound by the time this task
       // runs or else it's programmer error.
       DCHECK(proxy_task_runner_);
-      CHECK(false);

[Ken Rockot(use gerrit already), 2016/07/13 22:56:25]

Forgot to remove this on the last CL. :o

       std::unique_ptr<mojo::Message> passed_message(new mojo::Message);
       message->MoveTo(passed_message.get());
       proxy_task_runner_->PostTask(
           FROM_HERE,
           base::Bind(&ChannelAssociatedGroupController::AcceptOnProxyThread,
                      this, base::Passed(&passed_message)));
       return true;
     }
 
     // We do not expect to receive sync responses on the master endpoint thread.
     // If it's happening, it's a bug.
     DCHECK(!message->has_flag(mojo::Message::kFlagIsSync));
 
     bool result = false;
     {
       base::AutoUnlock unlocker(lock_);
       result = client->HandleIncomingMessage(message);
     }
 
     if (!result)
       RaiseError();
 
     return true;
   }
 
   void AcceptOnProxyThread(std::unique_ptr<mojo::Message> message) {
     DCHECK(proxy_task_runner_->BelongsToCurrentThread());
 
-    // TODO(rockot): Implement this.
-    NOTREACHED();
+    mojo::InterfaceId id = message->interface_id();
+    DCHECK(mojo::IsValidInterfaceId(id) && !mojo::IsMasterInterfaceId(id));
+
+    base::AutoLock locker(lock_);
+    bool inserted = false;
+    Endpoint* endpoint = FindOrInsertEndpoint(id, &inserted);
+
+    // The endpoint must have already been inserted by Accept(), and if the

[yzshen1, 2016/07/13 23:25:09]

It seems to be possible that |inserted| is true. Imagine:
1) a task to posted to handle a message on the proxy thread;
2) the corresponding local endpoint is closed;
3) the controller is shutdown on the IO thread, peer_closed is set for all
endpoints so the endpoints already marked as closed will be removed from the
registry.
4) the task of (1) is handled on the proxy thread.

WDYT?

[Ken Rockot(use gerrit already), 2016/07/14 00:39:09]

Ah your right, I had convinced myself that #2 was impossible, but it can happen.
Fixed.

+    // system is behaving as intended and consumers are well-behaved, it should
+    // be impossible for the endpoint to have been removed yet.
+    DCHECK(!inserted);
+
+    mojo::InterfaceEndpointClient* client = endpoint->client();
+    DCHECK(client);

[yzshen1, 2016/07/13 23:25:09]

The local endpoint could already be detached, right? (The associated interface
ptr/binding is already gone)

[Ken Rockot(use gerrit already), 2016/07/14 00:39:09]

Indeed. Done.

+    DCHECK(endpoint->task_runner()->BelongsToCurrentThread());
+
+    // TODO(rockot): Implement sync dispatch. For now, sync messages are
+    // unsupported here.
+    DCHECK(!message->has_flag(mojo::Message::kFlagIsSync));
+
+    bool result = false;
+    {
+      base::AutoUnlock unlocker(lock_);
+      result = client->HandleIncomingMessage(message.get());
+    }
+
+    if (!result)
+      RaiseError();
   }
 
   // mojo::PipeControlMessageHandlerDelegate:
   bool OnPeerAssociatedEndpointClosed(mojo::InterfaceId id) override {
     DCHECK(thread_checker_.CalledOnValidThread());
 
     if (mojo::IsMasterInterfaceId(id))
       return false;
 
     base::AutoLock locker(lock_);
@@ skipping 68 matching lines @@
   void set_connection_error_handler(const base::Closure& handler) {
     DCHECK(endpoint_client_);
     endpoint_client_->set_connection_error_handler(handler);
   }
 
   mojo::AssociatedGroup* associated_group() {
     DCHECK(controller_);
     return controller_->associated_group();
   }
 
+  ChannelAssociatedGroupController* controller() {
+    DCHECK(controller_);
+    return controller_.get();
+  }
+
   mojom::Bootstrap* operator->() {
     DCHECK(proxy_);
     return proxy_.get();
   }
 
  private:
   std::unique_ptr<mojom::BootstrapProxy> proxy_;
   scoped_refptr<ChannelAssociatedGroupController> controller_;
   std::unique_ptr<mojo::InterfaceEndpointClient> endpoint_client_;
 
@@ skipping 15 matching lines @@
   void set_connection_error_handler(const base::Closure& handler) {
     DCHECK(endpoint_client_);
     endpoint_client_->set_connection_error_handler(handler);
   }
 
   mojo::AssociatedGroup* associated_group() {
     DCHECK(controller_);
     return controller_->associated_group();
   }
 
+  ChannelAssociatedGroupController* controller() {
+    DCHECK(controller_);
+    return controller_.get();
+  }
+
   void Bind(mojo::ScopedMessagePipeHandle handle) {
     DCHECK(!controller_);
     controller_ =
         new ChannelAssociatedGroupController(false, std::move(handle));
     stub_.serialization_context()->group_controller = controller_;
 
     endpoint_client_.reset(new mojo::InterfaceEndpointClient(
         controller_->CreateLocalEndpointHandle(mojo::kMasterInterfaceId),
         &stub_,
         base::MakeUnique<typename mojom::Bootstrap::RequestValidator_>(),
@@ skipping 10 matching lines @@
 
 // MojoBootstrap for the server process. You should create the instance
 // using MojoBootstrap::Create().
 class MojoServerBootstrap : public MojoBootstrap {
  public:
   MojoServerBootstrap();
 
  private:
   // MojoBootstrap implementation.
   void Connect() override;
+
   mojo::AssociatedGroup* GetAssociatedGroup() override {
     return bootstrap_.associated_group();
   }
 
+  void SetProxyTaskRunner(
+      scoped_refptr<base::SingleThreadTaskRunner> task_runner) override {
+    bootstrap_.controller()->SetProxyTaskRunner(task_runner);
+  }
+
   void OnInitDone(int32_t peer_pid);
 
   BootstrapMasterProxy bootstrap_;
   IPC::mojom::ChannelAssociatedPtrInfo send_channel_;
   IPC::mojom::ChannelAssociatedRequest receive_channel_request_;
 
   DISALLOW_COPY_AND_ASSIGN(MojoServerBootstrap);
 };
 
 MojoServerBootstrap::MojoServerBootstrap() = default;
@@ skipping 38 matching lines @@
 
 // MojoBootstrap for client processes. You should create the instance
 // using MojoBootstrap::Create().
 class MojoClientBootstrap : public MojoBootstrap, public mojom::Bootstrap {
  public:
   MojoClientBootstrap();
 
  private:
   // MojoBootstrap implementation.
   void Connect() override;
+
   mojo::AssociatedGroup* GetAssociatedGroup() override {
     return binding_.associated_group();
   }
 
+  void SetProxyTaskRunner(
+      scoped_refptr<base::SingleThreadTaskRunner> task_runner) override {
+    binding_.controller()->SetProxyTaskRunner(task_runner);
+  }
+
   // mojom::Bootstrap implementation.
   void Init(mojom::ChannelAssociatedRequest receive_channel,
             mojom::ChannelAssociatedPtrInfo send_channel,
             int32_t peer_pid,
             const InitCallback& callback) override;
 
   BootstrapMasterBinding binding_;
 
   DISALLOW_COPY_AND_ASSIGN(MojoClientBootstrap);
 };
@@ skipping 66 matching lines @@
 
 bool MojoBootstrap::HasFailed() const {
   return state() == STATE_ERROR;
 }
 
 mojo::ScopedMessagePipeHandle MojoBootstrap::TakeHandle() {
   return std::move(handle_);
 }
 
 }  // namespace IPC