Change single-interface mojo bindings to use SequencedTaskRunner.

mojo/public/cpp/bindings/lib/multiplex_router.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/public/cpp/bindings/lib/multiplex_router.h"
 
 #include <stdint.h>
 
 #include <utility>
 
 #include "base/bind.h"
 #include "base/location.h"
 #include "base/macros.h"
 #include "base/memory/ptr_util.h"
-#include "base/single_thread_task_runner.h"
+#include "base/sequenced_task_runner.h"
 #include "base/stl_util.h"
+#include "base/threading/sequenced_task_runner_handle.h"
 #include "base/threading/thread_task_runner_handle.h"
 #include "mojo/public/cpp/bindings/associated_group.h"
 #include "mojo/public/cpp/bindings/interface_endpoint_client.h"
 #include "mojo/public/cpp/bindings/interface_endpoint_controller.h"
 #include "mojo/public/cpp/bindings/lib/may_auto_lock.h"
 #include "mojo/public/cpp/bindings/sync_handle_watcher.h"
 
 namespace mojo {
 namespace internal {
 
@@ skipping 36 matching lines @@
 
   const base::Optional<DisconnectReason>& disconnect_reason() const {
     return disconnect_reason_;
   }
   void set_disconnect_reason(
       const base::Optional<DisconnectReason>& disconnect_reason) {
     router_->AssertLockAcquired();
     disconnect_reason_ = disconnect_reason;
   }
 
-  base::SingleThreadTaskRunner* task_runner() const {
-    return task_runner_.get();
-  }
+  base::SequencedTaskRunner* task_runner() const { return task_runner_.get(); }
 
   InterfaceEndpointClient* client() const { return client_; }
 
   void AttachClient(InterfaceEndpointClient* client,
-                    scoped_refptr<base::SingleThreadTaskRunner> runner) {
+                    scoped_refptr<base::SequencedTaskRunner> runner) {
     router_->AssertLockAcquired();
     DCHECK(!client_);
     DCHECK(!closed_);
-    DCHECK(runner->BelongsToCurrentThread());
+    DCHECK(runner->RunsTasksOnCurrentThread());
 
     task_runner_ = std::move(runner);
     client_ = client;
   }
 
   // This method must be called on the same thread as the corresponding
   // AttachClient() call.
   void DetachClient() {
     router_->AssertLockAcquired();
     DCHECK(client_);
-    DCHECK(task_runner_->BelongsToCurrentThread());
+    DCHECK(task_runner_->RunsTasksOnCurrentThread());
     DCHECK(!closed_);
 
     task_runner_ = nullptr;
     client_ = nullptr;
     sync_watcher_.reset();
   }
 
   void SignalSyncMessageEvent() {
     router_->AssertLockAcquired();
     if (event_signalled_)
@@ skipping 20 matching lines @@
     DCHECK_EQ(MOJO_RESULT_OK, result);
     event_signalled_ = false;
   }
 
   // ---------------------------------------------------------------------------
   // The following public methods (i.e., InterfaceEndpointController
   // implementation) are called by the client on the same thread as the
   // AttachClient() call. They are called outside of the router's lock.
 
   bool SendMessage(Message* message) override {
-    DCHECK(task_runner_->BelongsToCurrentThread());
+    DCHECK(task_runner_->RunsTasksOnCurrentThread());
     message->set_interface_id(id_);
     return router_->connector_.Accept(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());
 
     EnsureSyncWatcherExists();
     return sync_watcher_->SyncWatch(should_stop);
   }
 
  private:
   friend class base::RefCounted<InterfaceEndpoint>;
 
   ~InterfaceEndpoint() override {
     router_->AssertLockAcquired();
 
     DCHECK(!client_);
     DCHECK(closed_);
     DCHECK(peer_closed_);
     DCHECK(!sync_watcher_);
   }
 
   void OnHandleReady(MojoResult result) {
-    DCHECK(task_runner_->BelongsToCurrentThread());
+    DCHECK(task_runner_->RunsTasksOnCurrentThread());
     scoped_refptr<InterfaceEndpoint> self_protector(this);
     scoped_refptr<MultiplexRouter> router_protector(router_);
 
     // Because we never close |sync_message_event_{sender,receiver}_| before
     // destruction or set a deadline, |result| should always be MOJO_RESULT_OK.
     DCHECK_EQ(MOJO_RESULT_OK, result);
     bool reset_sync_watcher = false;
     {
       MayAutoLock locker(router_->lock_.get());
 
       bool more_to_process = router_->ProcessFirstSyncMessageForEndpoint(id_);
 
       if (!more_to_process)
         ResetSyncMessageSignal();
 
       // Currently there are no queued sync messages and the peer has closed so
       // there won't be incoming sync messages in the future.
       reset_sync_watcher = !more_to_process && peer_closed_;
     }
     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;
 
     {
       MayAutoLock locker(router_->lock_.get());
       EnsureEventMessagePipeExists();
 
       auto iter = router_->sync_message_tasks_.find(id_);
       if (iter != router_->sync_message_tasks_.end() && !iter->second.empty())
         SignalSyncMessageEvent();
@@ skipping 25 matching lines @@
   // The following members are accessed under the router's lock.
 
   // Whether the endpoint has been closed.
   bool closed_;
   // Whether the peer endpoint has been closed.
   bool peer_closed_;
 
   base::Optional<DisconnectReason> disconnect_reason_;
 
   // The task runner on which |client_|'s methods can be called.
-  scoped_refptr<base::SingleThreadTaskRunner> task_runner_;
+  scoped_refptr<base::SequencedTaskRunner> task_runner_;
   // Not owned. It is null if no client is attached to this endpoint.
   InterfaceEndpointClient* client_;
 
   // A message pipe used as an event to signal that sync messages are available.
   // The message pipe handles are initialized under the router's lock and remain
   // unchanged afterwards. They may be accessed outside of the router's lock
   // later.
   ScopedMessagePipeHandle sync_message_event_sender_;
   ScopedMessagePipeHandle sync_message_event_receiver_;
   bool event_signalled_;
@@ skipping 35 matching lines @@
   Type type;
 
  private:
   explicit Task(Type in_type) : type(in_type) {}
 };
 
 MultiplexRouter::MultiplexRouter(
     ScopedMessagePipeHandle message_pipe,
     Config config,
     bool set_interface_id_namesapce_bit,
-    scoped_refptr<base::SingleThreadTaskRunner> runner)
+    scoped_refptr<base::SequencedTaskRunner> runner)
     : set_interface_id_namespace_bit_(set_interface_id_namesapce_bit),
       task_runner_(runner),
       header_validator_(nullptr),
       filters_(this),
       connector_(std::move(message_pipe),
                  config == MULTI_INTERFACE ? Connector::MULTI_THREADED_SEND
                                            : Connector::SINGLE_THREADED_SEND,
                  std::move(runner)),
       lock_(config == MULTI_INTERFACE ? new base::Lock : nullptr),
       control_message_handler_(this),
       control_message_proxy_(&connector_),
       next_interface_id_value_(1),
       posted_to_process_tasks_(false),
       encountered_error_(false),
       paused_(false),
       testing_mode_(false) {
-  DCHECK(task_runner_->BelongsToCurrentThread());
+  DCHECK(task_runner_->RunsTasksOnCurrentThread());
+  DCHECK(config == SINGLE_INTERFACE || base::ThreadTaskRunnerHandle::IsSet());
 
   if (config == SINGLE_INTERFACE_WITH_SYNC_METHODS ||
       config == MULTI_INTERFACE) {
     // Always participate in sync handle watching in multi-interface mode,
     // because even if it doesn't expect sync requests during sync handle
     // watching, it may still need to dispatch messages to associated endpoints
     // on a different thread.
     connector_.AllowWokenUpBySyncWatchOnSameThread();
   }
   connector_.set_incoming_receiver(&filters_);
@@ skipping 28 matching lines @@
       UpdateEndpointStateMayRemove(endpoint, ENDPOINT_CLOSED);
     } else {
       UpdateEndpointStateMayRemove(endpoint, PEER_ENDPOINT_CLOSED);
     }
   }
 
   DCHECK(endpoints_.empty());
 }
 
 void MultiplexRouter::SetMasterInterfaceName(const char* name) {
-  DCHECK(thread_checker_.CalledOnValidThread());
+  DCHECK(sequence_checker_.CalledOnValidSequence());
   header_validator_->SetDescription(
       std::string(name) + " [master] MessageHeaderValidator");
   control_message_handler_.SetDescription(
       std::string(name) + " [master] PipeControlMessageHandler");
   connector_.SetWatcherHeapProfilerTag(name);
 }
 
 void MultiplexRouter::CreateEndpointHandlePair(
     ScopedInterfaceEndpointHandle* local_endpoint,
     ScopedInterfaceEndpointHandle* remote_endpoint) {
@@ skipping 66 matching lines @@
     MayAutoUnlock unlocker(lock_.get());
     control_message_proxy_.NotifyPeerEndpointClosed(id, reason);
   }
 
   ProcessTasks(NO_DIRECT_CLIENT_CALLS, nullptr);
 }
 
 InterfaceEndpointController* MultiplexRouter::AttachEndpointClient(
     const ScopedInterfaceEndpointHandle& handle,
     InterfaceEndpointClient* client,
-    scoped_refptr<base::SingleThreadTaskRunner> runner) {
+    scoped_refptr<base::SequencedTaskRunner> runner) {
   const InterfaceId id = handle.id();
 
   DCHECK(IsValidInterfaceId(id));
   DCHECK(client);
 
   MayAutoLock locker(lock_.get());
   DCHECK(base::ContainsKey(endpoints_, id));
 
   InterfaceEndpoint* endpoint = endpoints_[id].get();
   endpoint->AttachClient(client, std::move(runner));
@@ skipping 12 matching lines @@
   DCHECK(IsValidInterfaceId(id));
 
   MayAutoLock locker(lock_.get());
   DCHECK(base::ContainsKey(endpoints_, id));
 
   InterfaceEndpoint* endpoint = endpoints_[id].get();
   endpoint->DetachClient();
 }
 
 void MultiplexRouter::RaiseError() {
-  if (task_runner_->BelongsToCurrentThread()) {
+  if (task_runner_->RunsTasksOnCurrentThread()) {
     connector_.RaiseError();
   } else {
     task_runner_->PostTask(FROM_HERE,
                            base::Bind(&MultiplexRouter::RaiseError, this));
   }
 }
 
 void MultiplexRouter::CloseMessagePipe() {
-  DCHECK(thread_checker_.CalledOnValidThread());
+  DCHECK(sequence_checker_.CalledOnValidSequence());
   connector_.CloseMessagePipe();
   // CloseMessagePipe() above won't trigger connection error handler.
   // Explicitly call OnPipeConnectionError() so that associated endpoints will
   // get notified.
   OnPipeConnectionError();
 }
 
 void MultiplexRouter::PauseIncomingMethodCallProcessing() {
-  DCHECK(thread_checker_.CalledOnValidThread());
+  DCHECK(sequence_checker_.CalledOnValidSequence());
   connector_.PauseIncomingMethodCallProcessing();
 
   MayAutoLock locker(lock_.get());
   paused_ = true;
 
   for (auto iter = endpoints_.begin(); iter != endpoints_.end(); ++iter)
     iter->second->ResetSyncMessageSignal();
 }
 
 void MultiplexRouter::ResumeIncomingMethodCallProcessing() {
-  DCHECK(thread_checker_.CalledOnValidThread());
+  DCHECK(sequence_checker_.CalledOnValidSequence());
   connector_.ResumeIncomingMethodCallProcessing();
 
   MayAutoLock locker(lock_.get());
   paused_ = false;
 
   for (auto iter = endpoints_.begin(); iter != endpoints_.end(); ++iter) {
     auto sync_iter = sync_message_tasks_.find(iter->first);
     if (sync_iter != sync_message_tasks_.end() && !sync_iter->second.empty())
       iter->second->SignalSyncMessageEvent();
   }
 
   ProcessTasks(NO_DIRECT_CLIENT_CALLS, nullptr);
 }
 
 bool MultiplexRouter::HasAssociatedEndpoints() const {
-  DCHECK(thread_checker_.CalledOnValidThread());
+  DCHECK(sequence_checker_.CalledOnValidSequence());
   MayAutoLock locker(lock_.get());
 
   if (endpoints_.size() > 1)
     return true;
   if (endpoints_.size() == 0)
     return false;
 
   return !base::ContainsKey(endpoints_, kMasterInterfaceId);
 }
 
 void MultiplexRouter::EnableTestingMode() {
-  DCHECK(thread_checker_.CalledOnValidThread());
+  DCHECK(sequence_checker_.CalledOnValidSequence());
   MayAutoLock locker(lock_.get());
 
   testing_mode_ = true;
   connector_.set_enforce_errors_from_incoming_receiver(false);
 }
 
 bool MultiplexRouter::Accept(Message* message) {
-  DCHECK(thread_checker_.CalledOnValidThread());
+  DCHECK(sequence_checker_.CalledOnValidSequence());
 
   scoped_refptr<MultiplexRouter> protector(this);
   MayAutoLock locker(lock_.get());
 
   DCHECK(!paused_);
 
   ClientCallBehavior client_call_behavior =
       connector_.during_sync_handle_watcher_callback()
           ? ALLOW_DIRECT_CLIENT_CALLS_FOR_SYNC_MESSAGES
           : ALLOW_DIRECT_CLIENT_CALLS;
@@ skipping 66 matching lines @@
   DCHECK(!endpoint->closed());
   UpdateEndpointStateMayRemove(endpoint, ENDPOINT_CLOSED);
 
   MayAutoUnlock unlocker(lock_.get());
   control_message_proxy_.NotifyPeerEndpointClosed(id, base::nullopt);
 
   return true;
 }
 
 void MultiplexRouter::OnPipeConnectionError() {
-  DCHECK(thread_checker_.CalledOnValidThread());
+  DCHECK(sequence_checker_.CalledOnValidSequence());
 
   scoped_refptr<MultiplexRouter> protector(this);
   MayAutoLock locker(lock_.get());
 
   encountered_error_ = true;
 
   for (auto iter = endpoints_.begin(); iter != endpoints_.end();) {
     InterfaceEndpoint* endpoint = iter->second.get();
     // Increment the iterator before calling UpdateEndpointStateMayRemove()
     // because it may remove the corresponding value from the map.
     ++iter;
 
     if (endpoint->client())
       tasks_.push_back(Task::CreateNotifyErrorTask(endpoint));
 
     UpdateEndpointStateMayRemove(endpoint, PEER_ENDPOINT_CLOSED);
   }
 
   ProcessTasks(connector_.during_sync_handle_watcher_callback()
                    ? ALLOW_DIRECT_CLIENT_CALLS_FOR_SYNC_MESSAGES
                    : ALLOW_DIRECT_CLIENT_CALLS,
                connector_.task_runner());
 }
 
 void MultiplexRouter::ProcessTasks(
     ClientCallBehavior client_call_behavior,
-    base::SingleThreadTaskRunner* current_task_runner) {
+    base::SequencedTaskRunner* current_task_runner) {
   AssertLockAcquired();
 
   if (posted_to_process_tasks_)
     return;
 
   while (!tasks_.empty() && !paused_) {
     std::unique_ptr<Task> task(std::move(tasks_.front()));
     tasks_.pop_front();
 
     InterfaceId id = kInvalidInterfaceId;
@@ skipping 59 matching lines @@
     sync_message_tasks_.erase(iter);
     return false;
   }
 
   return true;
 }
 
 bool MultiplexRouter::ProcessNotifyErrorTask(
     Task* task,
     ClientCallBehavior client_call_behavior,
-    base::SingleThreadTaskRunner* current_task_runner) {
-  DCHECK(!current_task_runner || current_task_runner->BelongsToCurrentThread());
+    base::SequencedTaskRunner* current_task_runner) {
+  DCHECK(!current_task_runner ||
+         current_task_runner->RunsTasksOnCurrentThread());
   DCHECK(!paused_);
 
   AssertLockAcquired();
   InterfaceEndpoint* endpoint = task->endpoint_to_notify.get();
   if (!endpoint->client())
     return true;
 
   if (client_call_behavior != ALLOW_DIRECT_CLIENT_CALLS ||
       endpoint->task_runner() != current_task_runner) {
     MaybePostToProcessTasks(endpoint->task_runner());
     return false;
   }
 
-  DCHECK(endpoint->task_runner()->BelongsToCurrentThread());
+  DCHECK(endpoint->task_runner()->RunsTasksOnCurrentThread());
 
   InterfaceEndpointClient* client = endpoint->client();
   base::Optional<DisconnectReason> disconnect_reason(
       endpoint->disconnect_reason());
 
   {
     // We must unlock before calling into |client| because it may call this
     // object within NotifyError(). Holding the lock will lead to deadlock.
     //
     // It is safe to call into |client| without the lock. Because |client| is
     // always accessed on the same thread, including DetachEndpointClient().
     MayAutoUnlock unlocker(lock_.get());
     client->NotifyError(disconnect_reason);
   }
   return true;
 }
 
 bool MultiplexRouter::ProcessIncomingMessage(
     Message* message,
     ClientCallBehavior client_call_behavior,
-    base::SingleThreadTaskRunner* current_task_runner) {
-  DCHECK(!current_task_runner || current_task_runner->BelongsToCurrentThread());
+    base::SequencedTaskRunner* current_task_runner) {
+  DCHECK(!current_task_runner ||
+         current_task_runner->RunsTasksOnCurrentThread());
   DCHECK(!paused_);
   DCHECK(message);
   AssertLockAcquired();
 
   if (message->IsNull()) {
     // This is a sync message and has been processed during sync handle
     // watching.
     return true;
   }
 
@@ skipping 35 matching lines @@
 
   if (!endpoint->client()) {
     // We need to wait until a client is attached in order to dispatch further
     // messages.
     return false;
   }
 
   bool can_direct_call;
   if (message->has_flag(Message::kFlagIsSync)) {
     can_direct_call = client_call_behavior != NO_DIRECT_CLIENT_CALLS &&
-                      endpoint->task_runner()->BelongsToCurrentThread();
+                      endpoint->task_runner()->RunsTasksOnCurrentThread();
   } else {
     can_direct_call = client_call_behavior == ALLOW_DIRECT_CLIENT_CALLS &&
                       endpoint->task_runner() == current_task_runner;
   }
 
   if (!can_direct_call) {
     MaybePostToProcessTasks(endpoint->task_runner());
     return false;
   }
 
-  DCHECK(endpoint->task_runner()->BelongsToCurrentThread());
+  DCHECK(endpoint->task_runner()->RunsTasksOnCurrentThread());
 
   InterfaceEndpointClient* client = endpoint->client();
   bool result = false;
   {
     // We must unlock before calling into |client| because it may call this
     // object within HandleIncomingMessage(). Holding the lock will lead to
     // deadlock.
     //
     // It is safe to call into |client| without the lock. Because |client| is
     // always accessed on the same thread, including DetachEndpointClient().
     MayAutoUnlock unlocker(lock_.get());
     result = client->HandleIncomingMessage(message);
   }
   if (!result)
     RaiseErrorInNonTestingMode();
 
   return true;
 }
 
 void MultiplexRouter::MaybePostToProcessTasks(
-    base::SingleThreadTaskRunner* task_runner) {
+    base::SequencedTaskRunner* task_runner) {
   AssertLockAcquired();
   if (posted_to_process_tasks_)
     return;
 
   posted_to_process_tasks_ = true;
   posted_to_task_runner_ = task_runner;
   task_runner->PostTask(
       FROM_HERE, base::Bind(&MultiplexRouter::LockAndCallProcessTasks, this));
 }
 
 void MultiplexRouter::LockAndCallProcessTasks() {
   // There is no need to hold a ref to this class in this case because this is
   // always called using base::Bind(), which holds a ref.
   MayAutoLock locker(lock_.get());
   posted_to_process_tasks_ = false;
-  scoped_refptr<base::SingleThreadTaskRunner> runner(
+  scoped_refptr<base::SequencedTaskRunner> runner(
       std::move(posted_to_task_runner_));
   ProcessTasks(ALLOW_DIRECT_CLIENT_CALLS, runner.get());
 }
 
 void MultiplexRouter::UpdateEndpointStateMayRemove(
     InterfaceEndpoint* endpoint,
     EndpointStateUpdateType type) {
   switch (type) {
     case ENDPOINT_CLOSED:
       endpoint->set_closed();
@@ skipping 39 matching lines @@
 
 void MultiplexRouter::AssertLockAcquired() {
 #if DCHECK_IS_ON()
   if (lock_)
     lock_->AssertAcquired();
 #endif
 }
 
 }  // namespace internal
 }  // namespace mojo