[mojo-edk] Prevent AcceptIncomingMessages flushing aggressively

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 602 matching lines @@
     scoped_refptr<NodeChannel> broker = GetBrokerChannel();
     if (!broker) {
       DVLOG(1) << "Dropping message for unknown peer: " << name;
       return;
     }
     broker->RequestIntroduction(name);
   }
 }
 
 void NodeController::AcceptIncomingMessages() {
-  while (incoming_messages_flag_) {
-    // TODO: We may need to be more careful to avoid starving the rest of the
-    // thread here. Revisit this if it turns out to be a problem. One
-    // alternative would be to schedule a task to continue pumping messages
-    // after flushing once.
+  {
+    base::AutoLock lock(messages_lock_);
+    if (!incoming_messages_.empty()) {
+      // libstdc++'s deque creates an internal buffer on construction, even when
+      // the size is 0. So avoid creating it until it is necessary.
+      std::queue<ports::ScopedMessage> messages;
+      std::swap(messages, incoming_messages_);
+      base::AutoUnlock unlock(messages_lock_);
 
-    messages_lock_.Acquire();
-    if (incoming_messages_.empty()) {
-      messages_lock_.Release();
-      break;
-    }
-    // libstdc++'s deque creates an internal buffer on construction, even when
-    // the size is 0. So avoid creating it until it is necessary.
-    std::queue<ports::ScopedMessage> messages;
-    std::swap(messages, incoming_messages_);
-    incoming_messages_flag_.Set(false);
-    messages_lock_.Release();
-
-    while (!messages.empty()) {
-      node_->AcceptMessage(std::move(messages.front()));
-      messages.pop();
+      while (!messages.empty()) {
+        node_->AcceptMessage(std::move(messages.front()));
+        messages.pop();
+      }
     }
   }
+
   AttemptShutdownIfRequested();
 }
 
 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);
   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_);
     incoming_messages_.emplace(std::move(message));
-    incoming_messages_flag_.Set(true);
   } else {
     SendPeerMessage(node, std::move(message));
   }
 }
 
 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());
@@ skipping 489 matching lines @@
     shutdown_callback_flag_.Set(false);
   }
 
   DCHECK(!callback.is_null());
 
   callback.Run();
 }
 
 }  // namespace edk
 }  // namespace mojo