[mojo] Implement pipe fusion API

mojo/edk/system/ports/node.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/ports/node.h"
 
 #include <string.h>
 
 #include "base/debug/alias.h"
 #include "base/logging.h"
@@ skipping 370 matching lines @@
     WillSendPort_Locked(port, destination_node_name, &data.new_port_name,
                         &data.new_port_descriptor);
     delegate_->ForwardMessage(
         destination_node_name,
         NewInternalMessage(destination_port_name,
                            EventType::kMergePort, data));
   }
   return OK;
 }
 
+int Node::MergeLocalPorts(const PortRef& port0_ref, const PortRef& port1_ref) {
+  Port* port0 = port0_ref.port();
+  Port* port1 = port1_ref.port();
+  int rv;
+  {
+    // |ports_lock_| must be held when acquiring overlapping port locks.
+    base::AutoLock ports_lock(ports_lock_);
+    base::AutoLock port0_lock(port0->lock);
+    base::AutoLock port1_lock(port1->lock);
+
+    DVLOG(1) << "Merging local ports " << port0_ref.name() << "@" << name_
+             << " and " << port1_ref.name() << "@" << name_;
+
+    if (port0->state != Port::kReceiving || port1->state != Port::kReceiving)
+      rv = ERROR_PORT_STATE_UNEXPECTED;
+    else
+      rv = MergePorts_Locked(port0_ref, port1_ref);
+  }
+
+  if (rv != OK) {
+    ClosePort(port0_ref);
+    ClosePort(port1_ref);
+  }
+
+  return rv;
+}
+
 int Node::LostConnectionToNode(const NodeName& node_name) {
   // We can no longer send events to the given node. We also can't expect any
   // PortAccepted events.
 
   DVLOG(1) << "Observing lost connection from node " << name_
            << " to node " << node_name;
 
   std::vector<PortRef> ports_to_notify;
 
   {
@@ skipping 355 matching lines @@
     if (port->state != Port::kReceiving) {
       close_new_port = true;
     } else {
       scoped_refptr<Port> new_port = GetPort_Locked(event.new_port_name);
       base::AutoLock new_port_lock(new_port->lock);
       DCHECK(new_port->state == Port::kReceiving);
 
       // Both ports are locked. Now all we have to do is swap their peer
       // information and set them up as proxies.
 
-      std::swap(port->peer_node_name, new_port->peer_node_name);
-      std::swap(port->peer_port_name, new_port->peer_port_name);
-      std::swap(port->peer_closed, new_port->peer_closed);
+      PortRef port0_ref(port_name, port);
+      PortRef port1_ref(event.new_port_name, new_port);
+      int rv = MergePorts_Locked(port0_ref, port1_ref);
+      if (rv == OK)
+        return rv;
 
-      port->state = Port::kBuffering;
-      if (port->peer_closed)
-        port->remove_proxy_on_last_message = true;
-
-      new_port->state = Port::kBuffering;
-      if (new_port->peer_closed)
-        new_port->remove_proxy_on_last_message = true;
-
-      int rv1 = BeginProxying_Locked(port.get(), port_name);
-      int rv2 = BeginProxying_Locked(new_port.get(), event.new_port_name);
-
-      if (rv1 == OK && rv2 == OK)
-        return OK;
-
-      // If either proxy failed to initialize (e.g. had undeliverable messages
-      // or ended up in a bad state somehow), we keep the system in a consistent
-      // state by undoing the peer swap and closing both merge ports.
-
-      std::swap(port->peer_node_name, new_port->peer_node_name);
-      std::swap(port->peer_port_name, new_port->peer_port_name);
-      port->state = Port::kReceiving;
-      new_port->state = Port::kReceiving;
       close_new_port = true;
       close_target_port = true;
     }
   }
 
   if (close_target_port) {
     PortRef target_port;
     rv = GetPort(port_name, &target_port);
     DCHECK(rv == OK);
 
@@ skipping 40 matching lines @@
 
 scoped_refptr<Port> Node::GetPort_Locked(const PortName& port_name) {
   ports_lock_.AssertAcquired();
   auto iter = ports_.find(port_name);
   if (iter == ports_.end())
     return nullptr;
 
   return iter->second;
 }
 
+int Node::MergePorts_Locked(const PortRef& port0_ref,
+                            const PortRef& port1_ref) {
+  Port* port0 = port0_ref.port();
+  Port* port1 = port1_ref.port();
+
+  ports_lock_.AssertAcquired();
+  port0->lock.AssertAcquired();
+  port1->lock.AssertAcquired();
+
+  CHECK(port0->state == Port::kReceiving);
+  CHECK(port1->state == Port::kReceiving);
+
+  // Ports cannot be merged with their own receiving peer!
+  if (port0->peer_node_name == name_ &&
+      port0->peer_port_name == port1_ref.name())
+    return ERROR_PORT_STATE_UNEXPECTED;
+
+  if (port1->peer_node_name == name_ &&
+      port1->peer_port_name == port0_ref.name())
+    return ERROR_PORT_STATE_UNEXPECTED;
+
+  // Only merge if both ports have never sent a message.
+  if (port0->next_sequence_num_to_send == kInitialSequenceNum &&
+      port1->next_sequence_num_to_send == kInitialSequenceNum) {
+    // Swap the ports' peer information and switch them both into buffering
+    // (eventually proxying) mode.
+
+    std::swap(port0->peer_node_name, port1->peer_node_name);
+    std::swap(port0->peer_port_name, port1->peer_port_name);
+    std::swap(port0->peer_closed, port1->peer_closed);
+
+    port0->state = Port::kBuffering;
+    if (port0->peer_closed)
+      port0->remove_proxy_on_last_message = true;
+
+    port1->state = Port::kBuffering;
+    if (port1->peer_closed)
+      port1->remove_proxy_on_last_message = true;
+
+    int rv1 = BeginProxying_Locked(port0, port0_ref.name());
+    int rv2 = BeginProxying_Locked(port1, port1_ref.name());
+
+    if (rv1 == OK && rv2 == OK) {
+      // If either merged port had a closed peer, its new peer needs to be
+      // informed of this.
+      if (port1->peer_closed) {
+        ObserveClosureEventData data;
+        data.last_sequence_num = port0->last_sequence_num_to_receive;
+        delegate_->ForwardMessage(
+            port0->peer_node_name,
+            NewInternalMessage(port0->peer_port_name,
+                               EventType::kObserveClosure, data));
+      }
+
+      if (port0->peer_closed) {
+        ObserveClosureEventData data;
+        data.last_sequence_num = port1->last_sequence_num_to_receive;
+        delegate_->ForwardMessage(
+            port1->peer_node_name,
+            NewInternalMessage(port1->peer_port_name,
+                               EventType::kObserveClosure, data));
+      }
+
+      return OK;
+    }
+
+    // If either proxy failed to initialize (e.g. had undeliverable messages
+    // or ended up in a bad state somehow), we keep the system in a consistent
+    // state by undoing the peer swap.
+    std::swap(port0->peer_node_name, port1->peer_node_name);
+    std::swap(port0->peer_port_name, port1->peer_port_name);
+    std::swap(port0->peer_closed, port1->peer_closed);
+    port0->remove_proxy_on_last_message = false;
+    port1->remove_proxy_on_last_message = false;
+    port0->state = Port::kReceiving;
+    port1->state = Port::kReceiving;
+  }
+
+  return ERROR_PORT_STATE_UNEXPECTED;
+}
+
 void Node::WillSendPort_Locked(Port* port,
                                const NodeName& to_node_name,
                                PortName* port_name,
                                PortDescriptor* port_descriptor) {
   ports_lock_.AssertAcquired();
   port->lock.AssertAcquired();
 
   PortName local_port_name = *port_name;
 
   PortName new_port_name;
@@ skipping 257 matching lines @@
 
   if (num_data_bytes)
     memcpy(header + 1, data, num_data_bytes);
 
   return message;
 }
 
 }  // namespace ports
 }  // namespace edk
 }  // namespace mojo