Rename TaskRunner::RunsTasksOnCurrentThread() in //extensions, //headless, //mojo

mojo/edk/system/node_controller.cc

 // Copyright 2016 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/node_controller.h"
 
 #include <algorithm>
 #include <limits>
 
 #include "base/bind.h"
@@ skipping 88 matching lines @@
 }
 
 // Used by NodeController to watch for shutdown. Since no IO can happen once
 // the IO thread is killed, the NodeController can cleanly drop all its peers
 // at that time.
 class ThreadDestructionObserver :
     public base::MessageLoop::DestructionObserver {
  public:
   static void Create(scoped_refptr<base::TaskRunner> task_runner,
                      const base::Closure& callback) {
-    if (task_runner->RunsTasksOnCurrentThread()) {
+    if (task_runner->RunsTasksInCurrentSequence()) {
       // Owns itself.
       new ThreadDestructionObserver(callback);
     } else {
       task_runner->PostTask(FROM_HERE,
                             base::Bind(&Create, task_runner, callback));
     }
   }
 
  private:
   explicit ThreadDestructionObserver(const base::Closure& callback)
@@ skipping 217 matching lines @@
   scoped_refptr<NodeChannel> peer = GetPeerChannel(source_node);
   if (peer)
     peer->NotifyBadMessage(error);
 }
 
 void NodeController::SendBrokerClientInvitationOnIOThread(
     base::ProcessHandle target_process,
     ConnectionParams connection_params,
     ports::NodeName temporary_node_name,
     const ProcessErrorCallback& process_error_callback) {
-  DCHECK(io_task_runner_->RunsTasksOnCurrentThread());
+  DCHECK(io_task_runner_->RunsTasksInCurrentSequence());
 
 #if !defined(OS_MACOSX) && !defined(OS_NACL)
   PlatformChannelPair node_channel;
   ScopedPlatformHandle server_handle = node_channel.PassServerHandle();
   // BrokerHost owns itself.
   BrokerHost* broker_host =
       new BrokerHost(target_process, connection_params.TakeChannelHandle());
   bool channel_ok = broker_host->SendChannel(node_channel.PassClientHandle());
 
 #if defined(OS_WIN)
@@ skipping 29 matching lines @@
 
   channel->SetRemoteNodeName(temporary_node_name);
   channel->SetRemoteProcessHandle(target_process);
   channel->Start();
 
   channel->AcceptChild(name_, temporary_node_name);
 }
 
 void NodeController::AcceptBrokerClientInvitationOnIOThread(
     ConnectionParams connection_params) {
-  DCHECK(io_task_runner_->RunsTasksOnCurrentThread());
+  DCHECK(io_task_runner_->RunsTasksInCurrentSequence());
 
   {
     base::AutoLock lock(parent_lock_);
     DCHECK(parent_name_ == ports::kInvalidNodeName);
 
     // At this point we don't know the parent's name, so we can't yet insert it
     // into our |peers_| map. That will happen as soon as we receive an
     // AcceptChild message from them.
     bootstrap_parent_channel_ =
         NodeChannel::Create(this, std::move(connection_params), io_task_runner_,
                             ProcessErrorCallback());
     // Prevent the parent pipe handle from being closed on shutdown. Pipe
     // closure is used by the parent to detect the child process has exited.
     // Relying on message pipes to be closed is not enough because the parent
     // may see the message pipe closure before the child is dead, causing the
     // child process to be unexpectedly SIGKILL'd.
     bootstrap_parent_channel_->LeakHandleOnShutdown();
   }
   bootstrap_parent_channel_->Start();
 }
 
 void NodeController::ConnectToPeerOnIOThread(uint64_t peer_connection_id,
                                              ConnectionParams connection_params,
                                              ports::PortRef port) {
-  DCHECK(io_task_runner_->RunsTasksOnCurrentThread());
+  DCHECK(io_task_runner_->RunsTasksInCurrentSequence());
 
   scoped_refptr<NodeChannel> channel = NodeChannel::Create(
       this, std::move(connection_params), io_task_runner_, {});
 
   ports::NodeName token;
   GenerateRandomName(&token);
   peer_connections_.emplace(token,
                             PeerConnection{channel, port, peer_connection_id});
   peer_connections_by_id_.emplace(peer_connection_id, token);
 
@@ skipping 42 matching lines @@
   {
     base::AutoLock lock(broker_lock_);
     broker_name = broker_name_;
   }
   return GetPeerChannel(broker_name);
 }
 
 void NodeController::AddPeer(const ports::NodeName& name,
                              scoped_refptr<NodeChannel> channel,
                              bool start_channel) {
-  DCHECK(io_task_runner_->RunsTasksOnCurrentThread());
+  DCHECK(io_task_runner_->RunsTasksInCurrentSequence());
 
   DCHECK(name != ports::kInvalidNodeName);
   DCHECK(channel);
 
   channel->SetRemoteNodeName(name);
 
   OutgoingMessageQueue pending_messages;
   {
     base::AutoLock lock(peers_lock_);
     if (peers_.find(name) != peers_.end()) {
@@ skipping 23 matching lines @@
 
   // Flush any queued message we need to deliver to this node.
   while (!pending_messages.empty()) {
     channel->PortsMessage(std::move(pending_messages.front()));
     pending_messages.pop();
   }
 }
 
 void NodeController::DropPeer(const ports::NodeName& name,
                               NodeChannel* channel) {
-  DCHECK(io_task_runner_->RunsTasksOnCurrentThread());
+  DCHECK(io_task_runner_->RunsTasksInCurrentSequence());
 
   {
     base::AutoLock lock(peers_lock_);
     auto it = peers_.find(name);
 
     if (it != peers_.end()) {
       ports::NodeName peer = it->first;
       peers_.erase(it);
       DVLOG(1) << "Dropped peer " << peer;
     }
@@ skipping 176 matching lines @@
     // done after AcceptIncomingMessages() otherwise a message might be missed.
     // Doing it here may result in at most two tasks existing at the same time;
     // this running one, and one pending in the task runner.
     incoming_messages_task_posted_ = false;
   }
 
   AcceptIncomingMessages();
 }
 
 void NodeController::DropAllPeers() {
-  DCHECK(io_task_runner_->RunsTasksOnCurrentThread());
+  DCHECK(io_task_runner_->RunsTasksInCurrentSequence());
 
   std::vector<scoped_refptr<NodeChannel>> all_peers;
   {
     base::AutoLock lock(parent_lock_);
     if (bootstrap_parent_channel_) {
       // |bootstrap_parent_channel_| isn't null'd here becuase we rely on its
       // existence to determine whether or not this is the root node. Once
       // bootstrap_parent_channel_->ShutDown() has been called,
       // |bootstrap_parent_channel_| is essentially a dead object and it doesn't
       // matter if it's deleted now or when |this| is deleted.
@@ skipping 92 matching lines @@
     observer->OnPortStatusChanged();
   } else {
     DVLOG(2) << "Ignoring status change for " << port.name() << " because it "
              << "doesn't have an observer.";
   }
 }
 
 void NodeController::OnAcceptChild(const ports::NodeName& from_node,
                                    const ports::NodeName& parent_name,
                                    const ports::NodeName& token) {
-  DCHECK(io_task_runner_->RunsTasksOnCurrentThread());
+  DCHECK(io_task_runner_->RunsTasksInCurrentSequence());
 
   scoped_refptr<NodeChannel> parent;
   {
     base::AutoLock lock(parent_lock_);
     if (bootstrap_parent_channel_ && parent_name_ == ports::kInvalidNodeName) {
       parent_name_ = parent_name;
       parent = bootstrap_parent_channel_;
     }
   }
 
   if (!parent) {
     DLOG(ERROR) << "Unexpected AcceptChild message from " << from_node;
     DropPeer(from_node, nullptr);
     return;
   }
 
   parent->SetRemoteNodeName(parent_name);
   parent->AcceptParent(token, name_);
 
   // NOTE: The child does not actually add its parent as a peer until
   // receiving an AcceptBrokerClient message from the broker. The parent
   // will request that said message be sent upon receiving AcceptParent.
 
   DVLOG(1) << "Child " << name_ << " accepting parent " << parent_name;
 }
 
 void NodeController::OnAcceptParent(const ports::NodeName& from_node,
                                     const ports::NodeName& token,
                                     const ports::NodeName& child_name) {
-  DCHECK(io_task_runner_->RunsTasksOnCurrentThread());
+  DCHECK(io_task_runner_->RunsTasksInCurrentSequence());
 
   auto it = pending_invitations_.find(from_node);
   if (it == pending_invitations_.end() || token != from_node) {
     DLOG(ERROR) << "Received unexpected AcceptParent message from "
                 << from_node;
     DropPeer(from_node, nullptr);
     return;
   }
 
   {
@@ skipping 184 matching lines @@
       message_queue.pop();
     }
   }
 #endif
 
   DVLOG(1) << "Child " << name_ << " accepted by broker " << broker_name;
 }
 
 void NodeController::OnPortsMessage(const ports::NodeName& from_node,
                                     Channel::MessagePtr channel_message) {
-  DCHECK(io_task_runner_->RunsTasksOnCurrentThread());
+  DCHECK(io_task_runner_->RunsTasksInCurrentSequence());
 
   void* data;
   size_t num_data_bytes, num_header_bytes, num_payload_bytes, num_ports_bytes;
   if (!ParsePortsMessage(channel_message.get(), &data, &num_data_bytes,
                          &num_header_bytes, &num_payload_bytes,
                          &num_ports_bytes)) {
     DropPeer(from_node, nullptr);
     return;
   }
 
   CHECK(channel_message);
   std::unique_ptr<PortsMessage> ports_message(
       new PortsMessage(num_header_bytes,
                        num_payload_bytes,
                        num_ports_bytes,
                        std::move(channel_message)));
   ports_message->set_source_node(from_node);
   node_->AcceptMessage(ports::ScopedMessage(ports_message.release()));
   AcceptIncomingMessages();
 }
 
 void NodeController::OnRequestPortMerge(
     const ports::NodeName& from_node,
     const ports::PortName& connector_port_name,
     const std::string& name) {
-  DCHECK(io_task_runner_->RunsTasksOnCurrentThread());
+  DCHECK(io_task_runner_->RunsTasksInCurrentSequence());
 
   DVLOG(2) << "Node " << name_ << " received RequestPortMerge for name " << name
            << " and port " << connector_port_name << "@" << from_node;
 
   ports::PortRef local_port;
   {
     base::AutoLock lock(reserved_ports_lock_);
     auto it = reserved_ports_.find(from_node);
     if (it == reserved_ports_.end()) {
       DVLOG(1) << "Ignoring port merge request from node " << from_node << ". "
@@ skipping 16 matching lines @@
 
   int rv = node_->MergePorts(local_port, from_node, connector_port_name);
   if (rv != ports::OK)
     DLOG(ERROR) << "MergePorts failed: " << rv;
 
   AcceptIncomingMessages();
 }
 
 void NodeController::OnRequestIntroduction(const ports::NodeName& from_node,
                                            const ports::NodeName& name) {
-  DCHECK(io_task_runner_->RunsTasksOnCurrentThread());
+  DCHECK(io_task_runner_->RunsTasksInCurrentSequence());
 
   scoped_refptr<NodeChannel> requestor = GetPeerChannel(from_node);
   if (from_node == name || name == ports::kInvalidNodeName || !requestor) {
     DLOG(ERROR) << "Rejecting invalid OnRequestIntroduction message from "
                 << from_node;
     DropPeer(from_node, nullptr);
     return;
   }
 
   scoped_refptr<NodeChannel> new_friend = GetPeerChannel(name);
   if (!new_friend) {
     // We don't know who they're talking about!
     requestor->Introduce(name, ScopedPlatformHandle());
   } else {
     PlatformChannelPair new_channel;
     requestor->Introduce(name, new_channel.PassServerHandle());
     new_friend->Introduce(from_node, new_channel.PassClientHandle());
   }
 }
 
 void NodeController::OnIntroduce(const ports::NodeName& from_node,
                                  const ports::NodeName& name,
                                  ScopedPlatformHandle channel_handle) {
-  DCHECK(io_task_runner_->RunsTasksOnCurrentThread());
+  DCHECK(io_task_runner_->RunsTasksInCurrentSequence());
 
   if (!channel_handle.is_valid()) {
     node_->LostConnectionToNode(name);
 
     DVLOG(1) << "Could not be introduced to peer " << name;
     base::AutoLock lock(peers_lock_);
     pending_peer_messages_.erase(name);
     return;
   }
 
@@ skipping 34 matching lines @@
            message->payload_size());
     iter.second->PortsMessage(std::move(peer_message));
   }
 }
 
 #if defined(OS_WIN) || (defined(OS_MACOSX) && !defined(OS_IOS))
 void NodeController::OnRelayPortsMessage(const ports::NodeName& from_node,
                                          base::ProcessHandle from_process,
                                          const ports::NodeName& destination,
                                          Channel::MessagePtr message) {
-  DCHECK(io_task_runner_->RunsTasksOnCurrentThread());
+  DCHECK(io_task_runner_->RunsTasksInCurrentSequence());
 
   if (GetBrokerChannel()) {
     // Only the broker should be asked to relay a message.
     LOG(ERROR) << "Non-broker refusing to relay message.";
     DropPeer(from_node, nullptr);
     return;
   }
 
   // The parent should always know which process this came from.
   DCHECK(from_process != base::kNullProcessHandle);
@@ skipping 57 matching lines @@
   }
 
   OnPortsMessage(source_node, std::move(message));
 }
 #endif
 
 void NodeController::OnAcceptPeer(const ports::NodeName& from_node,
                                   const ports::NodeName& token,
                                   const ports::NodeName& peer_name,
                                   const ports::PortName& port_name) {
-  DCHECK(io_task_runner_->RunsTasksOnCurrentThread());
+  DCHECK(io_task_runner_->RunsTasksInCurrentSequence());
 
   auto it = peer_connections_.find(from_node);
   if (it == peer_connections_.end()) {
     DLOG(ERROR) << "Received unexpected AcceptPeer message from " << from_node;
     DropPeer(from_node, nullptr);
     return;
   }
 
   scoped_refptr<NodeChannel> channel = std::move(it->second.channel);
   ports::PortRef local_port = it->second.local_port;
@@ skipping 16 matching lines @@
 
   // We need to choose one side to initiate the port merge. It doesn't matter
   // who does it as long as they don't both try. Simple solution: pick the one
   // with the "smaller" port name.
   if (local_port.name() < port_name)
     node()->MergePorts(local_port, peer_name, port_name);
 }
 
 void NodeController::OnChannelError(const ports::NodeName& from_node,
                                     NodeChannel* channel) {
-  if (io_task_runner_->RunsTasksOnCurrentThread()) {
+  if (io_task_runner_->RunsTasksInCurrentSequence()) {
     DropPeer(from_node, channel);
     // DropPeer may have caused local port closures, so be sure to process any
     // pending local messages.
     AcceptIncomingMessages();
   } else {
     io_task_runner_->PostTask(
         FROM_HERE,
         base::Bind(&NodeController::OnChannelError, base::Unretained(this),
                    from_node, channel));
   }
@@ skipping 76 matching lines @@
 NodeController::PeerConnection::~PeerConnection() = default;
 
 NodeController::PeerConnection& NodeController::PeerConnection::
 operator=(const PeerConnection& other) = default;
 
 NodeController::PeerConnection& NodeController::PeerConnection::
 operator=(PeerConnection&& other) = default;
 
 }  // namespace edk
 }  // namespace mojo