Change single-interface mojo bindings to use SequencedTaskRunner.

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 175 matching lines @@
     MarkClosedAndMaybeRemove(endpoint);
 
     base::AutoUnlock unlocker(lock_);
     if (!mojo::IsMasterInterfaceId(id) || reason)
       control_message_proxy_.NotifyPeerEndpointClosed(id, reason);
   }
 
   mojo::InterfaceEndpointController* AttachEndpointClient(
       const mojo::ScopedInterfaceEndpointHandle& handle,
       mojo::InterfaceEndpointClient* client,
-      scoped_refptr<base::SingleThreadTaskRunner> runner) override {
+      scoped_refptr<base::SequencedTaskRunner> runner) override {
     const mojo::InterfaceId id = handle.id();
 
     DCHECK(mojo::IsValidInterfaceId(id));
     DCHECK(client);
 
     base::AutoLock locker(lock_);
     DCHECK(ContainsKey(endpoints_, id));
 
     Endpoint* endpoint = endpoints_[id].get();
     endpoint->AttachClient(client, std::move(runner));
@@ skipping 63 matching lines @@
 
     const base::Optional<mojo::DisconnectReason>& disconnect_reason() const {
       return disconnect_reason_;
     }
 
     void set_disconnect_reason(
         const base::Optional<mojo::DisconnectReason>& disconnect_reason) {
       disconnect_reason_ = disconnect_reason;
     }
 
-    base::SingleThreadTaskRunner* task_runner() const {
+    base::SequencedTaskRunner* task_runner() const {
       return task_runner_.get();
     }
 
     mojo::InterfaceEndpointClient* client() const {
       controller_->lock_.AssertAcquired();
       return client_;
     }
 
     void AttachClient(mojo::InterfaceEndpointClient* client,
-                      scoped_refptr<base::SingleThreadTaskRunner> runner) {
+                      scoped_refptr<base::SequencedTaskRunner> runner) {
       controller_->lock_.AssertAcquired();
       DCHECK(!client_);
       DCHECK(!closed_);
-      DCHECK(runner->BelongsToCurrentThread());
+      DCHECK(runner->RunsTasksOnCurrentThread());
 
       task_runner_ = std::move(runner);
       client_ = client;
     }
 
     void DetachClient() {
       controller_->lock_.AssertAcquired();
       DCHECK(client_);
-      DCHECK(task_runner_->BelongsToCurrentThread());
+      DCHECK(task_runner_->RunsTasksOnCurrentThread());
       DCHECK(!closed_);
 
       task_runner_ = nullptr;
       client_ = nullptr;
       sync_watcher_.reset();
     }
 
     uint32_t EnqueueSyncMessage(mojo::Message message) {
       controller_->lock_.AssertAcquired();
       uint32_t id = GenerateSyncMessageId();
@@ skipping 12 matching lines @@
       controller_->lock_.AssertAcquired();
       if (sync_messages_.empty() || sync_messages_.front().first != id)
         return mojo::Message();
       mojo::Message message = std::move(sync_messages_.front().second);
       sync_messages_.pop();
       return message;
     }
 
     // mojo::InterfaceEndpointController:
     bool SendMessage(mojo::Message* message) override {
-      DCHECK(task_runner_->BelongsToCurrentThread());
+      DCHECK(task_runner_->RunsTasksOnCurrentThread());
       message->set_interface_id(id_);
       return controller_->SendMessage(message);
     }
 
     void AllowWokenUpBySyncWatchOnSameThread() override {
-      DCHECK(task_runner_->BelongsToCurrentThread());
+      DCHECK(task_runner_->RunsTasksOnCurrentThread());
 
       EnsureSyncWatcherExists();
       sync_watcher_->AllowWokenUpBySyncWatchOnSameThread();
     }
 
     bool SyncWatch(const bool* should_stop) override {
-      DCHECK(task_runner_->BelongsToCurrentThread());
+      DCHECK(task_runner_->RunsTasksOnCurrentThread());
 
       // It's not legal to make sync calls from the master endpoint's thread,
       // and in fact they must only happen from the proxy task runner.
       DCHECK(!controller_->task_runner_->BelongsToCurrentThread());
       DCHECK(controller_->proxy_task_runner_->BelongsToCurrentThread());
 
       EnsureSyncWatcherExists();
       return sync_watcher_->SyncWatch(should_stop);
     }
 
    private:
     friend class base::RefCountedThreadSafe<Endpoint>;
 
     ~Endpoint() override {
       controller_->lock_.AssertAcquired();
       DCHECK(!client_);
       DCHECK(closed_);
       DCHECK(peer_closed_);
       DCHECK(!sync_watcher_);
     }
 
     void OnSyncMessageEventHandleReady(MojoResult result) {
-      DCHECK(task_runner_->BelongsToCurrentThread());
+      DCHECK(task_runner_->RunsTasksOnCurrentThread());
 
       scoped_refptr<Endpoint> keepalive(this);
       scoped_refptr<AssociatedGroupController> controller_keepalive(
           controller_);
 
       bool reset_sync_watcher = false;
       {
         base::AutoLock locker(controller_->lock_);
         bool more_to_process = false;
         if (!sync_messages_.empty()) {
@@ skipping 24 matching lines @@
 
       if (reset_sync_watcher) {
         // If a SyncWatch() call (or multiple ones) of this interface endpoint
         // is on the call stack, resetting the sync watcher will allow it to
         // exit when the call stack unwinds to that frame.
         sync_watcher_.reset();
       }
     }
 
     void EnsureSyncWatcherExists() {
-      DCHECK(task_runner_->BelongsToCurrentThread());
+      DCHECK(task_runner_->RunsTasksOnCurrentThread());
       if (sync_watcher_)
         return;
 
       {
         base::AutoLock locker(controller_->lock_);
         EnsureSyncMessageEventExists();
         if (!sync_messages_.empty())
           SignalSyncMessageEvent();
       }
 
@@ skipping 16 matching lines @@
       return id;
     }
 
     ChannelAssociatedGroupController* const controller_;
     const mojo::InterfaceId id_;
 
     bool closed_ = false;
     bool peer_closed_ = false;
     base::Optional<mojo::DisconnectReason> disconnect_reason_;
     mojo::InterfaceEndpointClient* client_ = nullptr;
-    scoped_refptr<base::SingleThreadTaskRunner> task_runner_;
+    scoped_refptr<base::SequencedTaskRunner> task_runner_;
     std::unique_ptr<mojo::SyncHandleWatcher> sync_watcher_;
     std::unique_ptr<MojoEvent> sync_message_event_;
     std::queue<std::pair<uint32_t, mojo::Message>> sync_messages_;
     uint32_t next_sync_message_id_ = 0;
 
     DISALLOW_COPY_AND_ASSIGN(Endpoint);
   };
 
   class ControlMessageProxyThunk : public MessageReceiver {
    public:
@@ skipping 87 matching lines @@
       // Because a notification may in turn detach any endpoint, we have to
       // check each client again here.
       if (endpoint->client())
         NotifyEndpointOfError(endpoint.get(), false /* force_async */);
     }
   }
 
   void NotifyEndpointOfError(Endpoint* endpoint, bool force_async) {
     lock_.AssertAcquired();
     DCHECK(endpoint->task_runner() && endpoint->client());
-    if (endpoint->task_runner()->BelongsToCurrentThread() && !force_async) {
+    if (endpoint->task_runner()->RunsTasksOnCurrentThread() && !force_async) {
       mojo::InterfaceEndpointClient* client = endpoint->client();
       base::Optional<mojo::DisconnectReason> reason(
           endpoint->disconnect_reason());
 
       base::AutoUnlock unlocker(lock_);
       client->NotifyError(reason);
     } else {
       endpoint->task_runner()->PostTask(
           FROM_HERE,
           base::Bind(&ChannelAssociatedGroupController
                 ::NotifyEndpointOfErrorOnEndpointThread, this, endpoint->id(),
                 endpoint));
     }
   }
 
   void NotifyEndpointOfErrorOnEndpointThread(mojo::InterfaceId id,
                                              Endpoint* endpoint) {
     base::AutoLock locker(lock_);
     auto iter = endpoints_.find(id);
     if (iter == endpoints_.end() || iter->second.get() != endpoint)
       return;
     if (!endpoint->client())
       return;
 
-    DCHECK(endpoint->task_runner()->BelongsToCurrentThread());
+    DCHECK(endpoint->task_runner()->RunsTasksOnCurrentThread());
     NotifyEndpointOfError(endpoint, false /* force_async */);
   }
 
   void MarkClosedAndMaybeRemove(Endpoint* endpoint) {
     lock_.AssertAcquired();
     endpoint->set_closed();
     if (endpoint->closed() && endpoint->peer_closed())
       endpoints_.erase(endpoint->id());
   }
 
@@ skipping 27 matching lines @@
     if (mojo::PipeControlMessageHandler::IsPipeControlMessage(message))
       return control_message_handler_.Accept(message);
 
     mojo::InterfaceId id = message->interface_id();
     DCHECK(mojo::IsValidInterfaceId(id));
 
     base::AutoLock locker(lock_);
     Endpoint* endpoint = GetEndpointForDispatch(id, true /* create */);
     mojo::InterfaceEndpointClient* client =
         endpoint ? endpoint->client() : nullptr;
-    if (!client || !endpoint->task_runner()->BelongsToCurrentThread()) {
+    if (!client || !endpoint->task_runner()->RunsTasksOnCurrentThread()) {
       // 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_);
 
       if (message->has_flag(mojo::Message::kFlagIsSync)) {
         // Sync messages may need to be handled by the endpoint if it's blocking
@@ skipping 33 matching lines @@
 
     base::AutoLock locker(lock_);
     Endpoint* endpoint = GetEndpointForDispatch(id, false /* create */);
     if (!endpoint)
       return;
 
     mojo::InterfaceEndpointClient* client = endpoint->client();
     if (!client)
       return;
 
-    DCHECK(endpoint->task_runner()->BelongsToCurrentThread());
+    DCHECK(endpoint->task_runner()->RunsTasksOnCurrentThread());
 
     // Sync messages should never make their way to this method.
     DCHECK(!message.has_flag(mojo::Message::kFlagIsSync));
 
     bool result = false;
     {
       base::AutoUnlock unlocker(lock_);
       result = client->HandleIncomingMessage(&message);
     }
 
     if (!result)
       RaiseError();
   }
 
   void AcceptSyncMessage(mojo::InterfaceId interface_id, uint32_t message_id) {
     DCHECK(proxy_task_runner_->BelongsToCurrentThread());
 
     base::AutoLock locker(lock_);
     Endpoint* endpoint =
         GetEndpointForDispatch(interface_id, false /* create */);
     if (!endpoint)
       return;
 
-    DCHECK(endpoint->task_runner()->BelongsToCurrentThread());
+    DCHECK(endpoint->task_runner()->RunsTasksOnCurrentThread());
     mojo::Message message = endpoint->PopSyncMessage(message_id);
 
     // The message must have already been dequeued by the endpoint waking up
     // from a sync wait. Nothing to do.
     if (message.IsNull())
       return;
 
     mojo::InterfaceEndpointClient* client = endpoint->client();
     if (!client)
       return;
@@ skipping 154 matching lines @@
     Channel::Mode mode,
     Delegate* delegate,
     const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner) {
   return base::MakeUnique<MojoBootstrapImpl>(
       std::move(handle), delegate,
       new ChannelAssociatedGroupController(mode == Channel::MODE_SERVER,
                                            ipc_task_runner));
 }
 
 }  // namespace IPC