[mojo-edk] Process proxy disruptions locally.

mojo/edk/system/ports/ports_unittest.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 <stdio.h>
 #include <stdlib.h>
 #include <string.h>
 
 #include <map>
 #include <queue>
 #include <sstream>
 
 #include "base/logging.h"
 #include "base/rand_util.h"
+#include "mojo/edk/system/ports/event.h"
 #include "mojo/edk/system/ports/node.h"
 #include "mojo/edk/system/ports/node_delegate.h"
 #include "testing/gtest/include/gtest/gtest.h"
 
 namespace mojo {
 namespace edk {
 namespace ports {
 namespace test {
 
 namespace {
@@ skipping 82 matching lines @@
     task_queue.pop();
 
     Node* node = GetNode(task->node_name);
     if (node)
       node->AcceptMessage(std::move(task->message));
 
     delete task;
   }
 }
 
+void PumpUntilTask(EventType type) {
+  while (!task_queue.empty()) {
+    Task* task = task_queue.top();
+
+    const EventHeader* header = GetEventHeader(*task->message);
+    if (header->type == type)
+      return;
+
+    task_queue.pop();
+
+    Node* node = GetNode(task->node_name);
+    if (node)
+      node->AcceptMessage(std::move(task->message));
+
+    delete task;
+  }
+}
+
 void DiscardPendingTasks() {
   while (!task_queue.empty()) {
     Task* task = task_queue.top();
     task_queue.pop();
     delete task;
   }
 }
 
 int SendStringMessage(Node* node, const PortRef& port, const std::string& s) {
   size_t size = s.size() + 1;
@@ skipping 66 matching lines @@
       return;
     }
     DVLOG(1) << "ForwardMessage from node "
              << node_name_ << " to " << node_name;
     task_queue.push(new Task(node_name, std::move(message)));
   }
 
   void BroadcastMessage(ScopedMessage message) override {
     for (size_t i = 0; i < kMaxNodes; ++i) {
       Node* node = node_map[i];
-      if (node) {
+      // Broadcast doesn't deliver to the local node.
+      if (node && node != GetNode(node_name_)) {
         // NOTE: We only need to support broadcast of events, which have no
         // payload or ports bytes.
         ScopedMessage new_message(
             new TestMessage(message->num_header_bytes(), 0, 0));
         memcpy(new_message->mutable_header_bytes(), message->header_bytes(),
                message->num_header_bytes());
         node->AcceptMessage(std::move(new_message));
       }
     }
   }
@@ skipping 328 matching lines @@
 
   EXPECT_EQ(OK, node0.ClosePort(A));
   EXPECT_EQ(OK, node1.ClosePort(B));
   EXPECT_EQ(OK, node1.ClosePort(C));
   EXPECT_EQ(OK, node0.ClosePort(E));
 
   EXPECT_TRUE(node0.CanShutdownCleanly(false));
   EXPECT_TRUE(node1.CanShutdownCleanly(false));
 }
 
+TEST_F(PortsTest, LostConnectionToNodeWithLocalProxy) {
+  // Tests that a proxy gets cleaned up when its direct peer lives on a lost
+  // node and it's predecessor lives on the same node.
+
+  NodeName node0_name(0, 1);
+  TestNodeDelegate node0_delegate(node0_name);
+  Node node0(node0_name, &node0_delegate);
+  node_map[0] = &node0;
+
+  NodeName node1_name(1, 1);
+  TestNodeDelegate node1_delegate(node1_name);
+  Node node1(node1_name, &node1_delegate);
+  node_map[1] = &node1;
+
+  node1_delegate.set_save_messages(true);
+
+  // Create A-B spanning nodes 0 and 1.
+  PortRef A, B;
+  EXPECT_EQ(OK, node0.CreateUninitializedPort(&A));
+  EXPECT_EQ(OK, node1.CreateUninitializedPort(&B));
+  EXPECT_EQ(OK, node0.InitializePort(A, node1_name, B.name()));
+  EXPECT_EQ(OK, node1.InitializePort(B, node0_name, A.name()));
+
+  // Create C-D and send D over A to node 1.
+  PortRef C, D;
+  EXPECT_EQ(OK, node0.CreatePortPair(&C, &D));
+  EXPECT_EQ(OK, SendStringMessageWithPort(&node0, A, ".", D));
+
+  // Pump tasks until the start of port collapse for port D, which should become
+  // a proxy.
+  PumpUntilTask(EventType::kObserveProxy);
+
+  ScopedMessage message;
+  ASSERT_TRUE(node1_delegate.GetSavedMessage(&message));
+  ASSERT_EQ(1u, message->num_ports());
+
+  PortRef E;
+  EXPECT_EQ(OK, node1.GetPort(message->ports()[0], &E));
+
+  EXPECT_EQ(OK, node0.LostConnectionToNode(node1_name));
+  PumpTasks();
+
+  // Port C should have detected peer closure.
+  PortStatus status;
+  EXPECT_EQ(OK, node0.GetStatus(C, &status));
+  EXPECT_TRUE(status.peer_closed);
+
+  EXPECT_EQ(OK, node0.ClosePort(A));
+  EXPECT_EQ(OK, node1.ClosePort(B));
+  EXPECT_EQ(OK, node0.ClosePort(C));
+  EXPECT_EQ(OK, node1.ClosePort(E));
+
+  EXPECT_TRUE(node0.CanShutdownCleanly(false));
+  EXPECT_TRUE(node1.CanShutdownCleanly(false));
+}
+
 TEST_F(PortsTest, GetMessage1) {
   NodeName node0_name(0, 1);
   TestNodeDelegate node0_delegate(node0_name);
   Node node0(node0_name, &node0_delegate);
   node_map[0] = &node0;
 
   PortRef a0, a1;
   EXPECT_EQ(OK, node0.CreatePortPair(&a0, &a1));
 
   ScopedMessage message;
@@ skipping 872 matching lines @@
 
   // Expect everything to have gone away.
   EXPECT_TRUE(node0.CanShutdownCleanly(false));
   EXPECT_TRUE(node1.CanShutdownCleanly(false));
 }
 
 }  // namespace test
 }  // namespace ports
 }  // namespace edk
 }  // namespace mojo