[mojo-edk] Broadcast surprise port disruptions

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>
@@ skipping 70 matching lines @@
   ScopedMessage message;
   uint64_t priority;
 };
 
 struct TaskComparator {
   bool operator()(const Task* a, const Task* b) {
     return a->priority < b->priority;
   }
 };
 
+const size_t kMaxNodes = 3;
+
 std::priority_queue<Task*, std::vector<Task*>, TaskComparator> task_queue;
-Node* node_map[2];
+Node* node_map[kMaxNodes];
 
 Node* GetNode(const NodeName& name) {
   return node_map[name.v1];
 }
 
 void SetNode(const NodeName& name, Node* node) {
   node_map[name.v1] = node;
 }
 
 void PumpTasks() {
   while (!task_queue.empty()) {
     Task* task = task_queue.top();
     task_queue.pop();
 
     Node* node = GetNode(task->node_name);
-    node->AcceptMessage(std::move(task->message));
+    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;
@@ skipping 68 matching lines @@
       DVLOG(1) << "Dropping ForwardMessage from node "
                << node_name_ << " to " << node_name;
       ClosePortsInMessage(GetNode(node_name), message.get());
       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) {
+        // 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));
+      }
+    }
+  }
+
   void PortStatusChanged(const PortRef& port) override {
     DVLOG(1) << "PortStatusChanged for " << port.name() << "@" << node_name_;
     if (!read_messages_)
       return;
     Node* node = GetNode(node_name_);
     for (;;) {
       ScopedMessage message;
       int rv = node->GetMessage(port, &message);
       EXPECT_TRUE(rv == OK || rv == ERROR_PORT_PEER_CLOSED);
       if (rv == ERROR_PORT_PEER_CLOSED || !message)
@@ skipping 25 matching lines @@
 
   std::queue<ScopedMessage> saved_messages_;
   NodeName node_name_;
   bool drop_messages_;
   bool read_messages_;
   bool save_messages_;
 };
 
 class PortsTest : public testing::Test {
  public:
-  PortsTest() {
-    SetNode(NodeName(0, 1), nullptr);
-    SetNode(NodeName(1, 1), nullptr);
-  }
-
-  ~PortsTest() override {
+  void SetUp() override {
     DiscardPendingTasks();
     SetNode(NodeName(0, 1), nullptr);
     SetNode(NodeName(1, 1), nullptr);
+    SetNode(NodeName(2, 1), nullptr);
   }
 };
 
 }  // namespace
 
 TEST_F(PortsTest, Basic1) {
   NodeName node0_name(0, 1);
   TestNodeDelegate node0_delegate(node0_name);
   Node node0(node0_name, &node0_delegate);
   SetNode(node0_name, &node0);
@@ skipping 193 matching lines @@
   ClosePortsInMessage(&node1, message.get());
 
   EXPECT_EQ(OK, node1.ClosePort(x1));
 
   PumpTasks();
 
   EXPECT_TRUE(node0.CanShutdownCleanly(false));
   EXPECT_TRUE(node1.CanShutdownCleanly(false));
 }
 
+TEST_F(PortsTest, LostConnectionToNodeWithSecondaryProxy) {
+  // Tests that a proxy gets cleaned up when its indirect peer lives on a lost
+  // 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;
+
+  NodeName node2_name(2, 1);
+  TestNodeDelegate node2_delegate(node2_name);
+  Node node2(node2_name, &node2_delegate);
+  node_map[2] = &node2;
+
+  node1_delegate.set_save_messages(true);
+
+  // Create A-B spanning nodes 0 and 1 and C-D spanning 1 and 2.
+  PortRef A, B, C, D;
+  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()));
+  EXPECT_EQ(OK, node1.CreateUninitializedPort(&C));
+  EXPECT_EQ(OK, node2.CreateUninitializedPort(&D));
+  EXPECT_EQ(OK, node1.InitializePort(C, node2_name, D.name()));
+  EXPECT_EQ(OK, node2.InitializePort(D, node1_name, C.name()));
+
+  // Create E-F and send F over A to node 1.
+  PortRef E, F;
+  EXPECT_EQ(OK, node0.CreatePortPair(&E, &F));
+  EXPECT_EQ(OK, SendStringMessageWithPort(&node0, A, ".", F));
+
+  PumpTasks();
+
+  ScopedMessage message;
+  ASSERT_TRUE(node1_delegate.GetSavedMessage(&message));
+  ASSERT_EQ(1u, message->num_ports());
+
+  EXPECT_EQ(OK, node1.GetPort(message->ports()[0], &F));
+
+  // Send F over C to node 2 and then simulate node 2 loss from node 1. Node 1
+  // will trivially become aware of the loss, and this test verifies that the
+  // port A on node 0 will eventually also become aware of it.
+
+  EXPECT_EQ(OK, SendStringMessageWithPort(&node1, C, ".", F));
+
+  node_map[2] = nullptr;
+  EXPECT_EQ(OK, node1.LostConnectionToNode(node2_name));
+
+  PumpTasks();
+
+  // Port F should be gone.
+  EXPECT_EQ(ERROR_PORT_UNKNOWN, node1.GetPort(F.name(), &F));
+
+  // Port E should have detected peer closure despite the fact that there is
+  // no longer a continuous route from F to E over which the event could travel.
+  PortStatus status;
+  EXPECT_EQ(OK, node0.GetStatus(E, &status));
+  EXPECT_TRUE(status.peer_closed);
+
+  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, 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