[mojo-edk] Post a task to process local messages.

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"
 #include "base/location.h"
 #include "base/logging.h"
 #include "base/macros.h"
 #include "base/message_loop/message_loop.h"
 #include "base/metrics/histogram_macros.h"
 #include "base/process/process_handle.h"
 #include "base/rand_util.h"
 #include "base/time/time.h"
 #include "base/timer/elapsed_timer.h"
 #include "mojo/edk/embedder/embedder_internal.h"
 #include "mojo/edk/embedder/platform_channel_pair.h"
 #include "mojo/edk/system/broker.h"
 #include "mojo/edk/system/broker_host.h"
 #include "mojo/edk/system/core.h"
 #include "mojo/edk/system/ports_message.h"
+#include "mojo/edk/system/request_context.h"
 
 #if defined(OS_MACOSX) && !defined(OS_IOS)
 #include "mojo/edk/system/mach_port_relay.h"
 #endif
 
 #if !defined(OS_NACL)
 #include "crypto/random.h"
 #endif
 
 namespace mojo {
@@ skipping 603 matching lines @@
     messages_lock_.Release();
 
     while (!messages.empty()) {
       node_->AcceptMessage(std::move(messages.front()));
       messages.pop();
     }
   }
   AttemptShutdownIfRequested();
 }
 
+void NodeController::ProcessIncomingMessages() {
+  RequestContext request_context(RequestContext::Source::SYSTEM);
+  AcceptIncomingMessages();
+}
+
 void NodeController::DropAllPeers() {
   DCHECK(io_task_runner_->RunsTasksOnCurrentThread());
 
   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,
@@ skipping 27 matching lines @@
 }
 
 void NodeController::AllocMessage(size_t num_header_bytes,
                                   ports::ScopedMessage* message) {
   message->reset(new PortsMessage(num_header_bytes, 0, 0, nullptr));
 }
 
 void NodeController::ForwardMessage(const ports::NodeName& node,
                                     ports::ScopedMessage message) {
   DCHECK(message);
+  bool schedule_pump_task = false;
   if (node == name_) {
     // NOTE: We need to avoid re-entering the Node instance within
     // ForwardMessage. Because ForwardMessage is only ever called
     // (synchronously) in response to Node's ClosePort, SendMessage, or
     // AcceptMessage, we flush the queue after calling any of those methods.
     base::AutoLock lock(messages_lock_);
+    schedule_pump_task = incoming_messages_.empty();
     incoming_messages_.emplace(std::move(message));
     incoming_messages_flag_.Set(true);
   } else {
     SendPeerMessage(node, std::move(message));
   }
+
+  // |io_task_runner_| may be null in tests or processes that don't require
+  // multi-process Mojo.
+  if (schedule_pump_task && io_task_runner_) {
+    // Normally, the queue is processed after the action that added the local
+    // message is done (i.e. SendMessage, ClosePort, etc). However, it's also
+    // possible for a local message to be added as a result of a remote message,
+    // and OnChannelMessage() doesn't process this queue (although
+    // OnPortsMessage() does). There may also be other code paths, now or added
+    // in the future, which cause local messages to be added but don't process
+    // this message queue.
+    //
+    // Instead of adding a call to AcceptIncomingMessages() on every possible
+    // code path, post a task to the IO thread to process the queue. If the
+    // current call stack processes the queue, this may end up doing nothing.
+    io_task_runner_->PostTask(
+        FROM_HERE,
+        base::Bind(&NodeController::ProcessIncomingMessages,
+                   base::Unretained(this)));
+  }
 }
 
 void NodeController::BroadcastMessage(ports::ScopedMessage message) {
   CHECK_EQ(message->num_ports(), 0u);
   Channel::MessagePtr channel_message =
       static_cast<PortsMessage*>(message.get())->TakeChannelMessage();
   CHECK(!channel_message->has_handles());
 
   scoped_refptr<NodeChannel> broker = GetBrokerChannel();
   if (broker)
@@ skipping 486 matching lines @@
     shutdown_callback_flag_.Set(false);
   }
 
   DCHECK(!callback.is_null());
 
   callback.Run();
 }
 
 }  // namespace edk
 }  // namespace mojo