[mojo-edk] Ensure there is only one ProcessIncomingMessages() task posted to the IO thread.

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 633 matching lines @@
     while (!messages.empty()) {
       node_->AcceptMessage(std::move(messages.front()));
       messages.pop();
     }
   }
   AttemptShutdownIfRequested();
 }
 
 void NodeController::ProcessIncomingMessages() {
   RequestContext request_context(RequestContext::Source::SYSTEM);
+
+  {
+    base::AutoLock lock(messages_lock_);
+    // Allow a new incoming messages processing task to be posted. This can't be
+    // 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());
 
   std::vector<scoped_refptr<NodeChannel>> all_peers;
   {
     base::AutoLock lock(parent_lock_);
     if (bootstrap_parent_channel_) {
@@ skipping 37 matching lines @@
 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();
+    // |io_task_runner_| may be null in tests or processes that don't require
+    // multi-process Mojo.
+    schedule_pump_task = incoming_messages_.empty() && io_task_runner_ &&
+        !incoming_messages_task_posted_;
+    incoming_messages_task_posted_ |= schedule_pump_task;
     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_) {
+  if (schedule_pump_task) {
     // 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
@@ skipping 506 matching lines @@
     shutdown_callback_flag_.Set(false);
   }
 
   DCHECK(!callback.is_null());
 
   callback.Run();
 }
 
 }  // namespace edk
 }  // namespace mojo