Revert of [mojo-edk] Simplify multiprocess pipe bootstrap

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/logging.h"
 #include "base/memory/ref_counted.h"
@@ skipping 336 matching lines @@
           header->port_name,
           *GetEventData<ObserveProxyEventData>(*message));
     case EventType::kObserveProxyAck:
       return OnObserveProxyAck(
           header->port_name,
           GetEventData<ObserveProxyAckEventData>(*message)->last_sequence_num);
     case EventType::kObserveClosure:
       return OnObserveClosure(
           header->port_name,
           GetEventData<ObserveClosureEventData>(*message)->last_sequence_num);
-    case EventType::kMergePort:
-      return OnMergePort(header->port_name,
-                         *GetEventData<MergePortEventData>(*message));
   }
   return OOPS(ERROR_NOT_IMPLEMENTED);
 }
 
-int Node::MergePorts(const PortRef& port_ref,
-                     const NodeName& destination_node_name,
-                     const PortName& destination_port_name) {
-  Port* port = port_ref.port();
-  {
-    // |ports_lock_| must be held for WillSendPort_Locked below.
-    base::AutoLock ports_lock(ports_lock_);
-    base::AutoLock lock(port->lock);
-
-    DVLOG(1) << "Sending MergePort from " << port_ref.name() << "@" << name_
-             << " to " << destination_port_name << "@" << destination_node_name;
-
-    // Send the port-to-merge over to the destination node so it can be merged
-    // into the port cycle atomically there.
-    MergePortEventData data;
-    data.new_port_name = port_ref.name();
-    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::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 131 matching lines @@
   {
     // We must hold |ports_lock_| before grabbing the port lock because
     // ForwardMessages_Locked requires it to be held.
     base::AutoLock ports_lock(ports_lock_);
     base::AutoLock lock(port->lock);
 
     DVLOG(2) << "PortAccepted at " << port_name << "@" << name_
              << " pointing to "
              << port->peer_port_name << "@" << port->peer_node_name;
 
-    return BeginProxying_Locked(port.get(), port_name);
+    if (port->state != Port::kBuffering)
+      return OOPS(ERROR_PORT_STATE_UNEXPECTED);
+
+    port->state = Port::kProxying;
+
+    int rv = ForwardMessages_Locked(port.get(), port_name);
+    if (rv != OK)
+      return rv;
+
+    // We may have observed closure before receiving PortAccepted. In that
+    // case, we can advance to removing the proxy without sending out an
+    // ObserveProxy message. We already know the last expected message, etc.
+
+    if (port->remove_proxy_on_last_message) {
+      MaybeRemoveProxy_Locked(port.get(), port_name);
+
+      // Make sure we propagate closure to our current peer.
+      ObserveClosureEventData data;
+      data.last_sequence_num = port->last_sequence_num_to_receive;
+      delegate_->ForwardMessage(
+          port->peer_node_name,
+          NewInternalMessage(port->peer_port_name,
+                             EventType::kObserveClosure, data));
+    } else {
+      InitiateProxyRemoval_Locked(port.get(), port_name);
+    }
   }
+  return OK;
 }
 
 int Node::OnObserveProxy(const PortName& port_name,
                          const ObserveProxyEventData& event) {
   // The port may have already been closed locally, in which case the
   // ObserveClosure message will contain the last_sequence_num field.
   // We can then silently ignore this message.
   scoped_refptr<Port> port = GetPort(port_name);
   if (!port) {
     DVLOG(1) << "ObserveProxy: " << port_name << "@" << name_ << " not found";
@@ skipping 164 matching lines @@
         NewInternalMessage(port->peer_port_name,
                            EventType::kObserveClosure, forwarded_data));
   }
   if (notify_delegate) {
     PortRef port_ref(port_name, port);
     delegate_->PortStatusChanged(port_ref);
   }
   return OK;
 }
 
-int Node::OnMergePort(const PortName& port_name,
-                      const MergePortEventData& event) {
-  scoped_refptr<Port> port = GetPort(port_name);
-  if (!port)
-    return ERROR_PORT_UNKNOWN;
-
-  DVLOG(1) << "MergePort at " << port_name << "@" << name_ << " (state="
-           << port->state << ") merging with proxy " << event.new_port_name
-           << "@" << name_ << " pointing to "
-           << event.new_port_descriptor.peer_port_name << "@"
-           << event.new_port_descriptor.peer_node_name << " referred by "
-           << event.new_port_descriptor.referring_port_name << "@"
-           << event.new_port_descriptor.referring_node_name;
-
-  bool close_target_port = false;
-  bool close_new_port = false;
-
-  // Accept the new port. This is now the receiving end of the other port cycle
-  // to be merged with ours.
-  int rv = AcceptPort(event.new_port_name, event.new_port_descriptor);
-  if (rv != OK) {
-    close_target_port = true;
-  } else {
-    // BeginProxying_Locked may call MaybeRemoveProxy_Locked, which in turn
-    // needs to hold |ports_lock_|. We also acquire multiple port locks within.
-    base::AutoLock ports_lock(ports_lock_);
-    base::AutoLock lock(port->lock);
-
-    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);
-
-      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);
-
-    ClosePort(target_port);
-  }
-
-  if (close_new_port) {
-    PortRef new_port;
-    rv = GetPort(event.new_port_name, &new_port);
-    DCHECK(rv == OK);
-
-    ClosePort(new_port);
-  }
-
-  return ERROR_PORT_STATE_UNEXPECTED;
-}
-
 int Node::AddPortWithName(const PortName& port_name,
                           const scoped_refptr<Port>& port) {
   base::AutoLock lock(ports_lock_);
 
   if (!ports_.insert(std::make_pair(port_name, port)).second)
     return OOPS(ERROR_PORT_EXISTS);  // Suggests a bad UUID generator.
 
   DVLOG(2) << "Created port " << port_name << "@" << name_;
   return OK;
 }
@@ skipping 167 matching lines @@
            << GetEventData<UserEventData>(*message)->sequence_num
            << " [ports=" << ports_buf.str() << "]"
            << " from " << port_name << "@" << name_
            << " to " << port->peer_port_name << "@" << port->peer_node_name;
 #endif
 
   GetMutableEventHeader(message)->port_name = port->peer_port_name;
   return OK;
 }
 
-int Node::BeginProxying_Locked(Port* port, const PortName& port_name) {
-  ports_lock_.AssertAcquired();
-  port->lock.AssertAcquired();
-
-  if (port->state != Port::kBuffering)
-    return OOPS(ERROR_PORT_STATE_UNEXPECTED);
-
-  port->state = Port::kProxying;
-
-  int rv = ForwardMessages_Locked(port, port_name);
-  if (rv != OK)
-    return rv;
-
-  // We may have observed closure while buffering. In that case, we can advance
-  // to removing the proxy without sending out an ObserveProxy message. We
-  // already know the last expected message, etc.
-
-  if (port->remove_proxy_on_last_message) {
-    MaybeRemoveProxy_Locked(port, port_name);
-
-    // Make sure we propagate closure to our current peer.
-    ObserveClosureEventData data;
-    data.last_sequence_num = port->last_sequence_num_to_receive;
-    delegate_->ForwardMessage(
-        port->peer_node_name,
-        NewInternalMessage(port->peer_port_name,
-                           EventType::kObserveClosure, data));
-  } else {
-    InitiateProxyRemoval_Locked(port, port_name);
-  }
-
-  return OK;
-}
-
 int Node::ForwardMessages_Locked(Port* port, const PortName &port_name) {
   ports_lock_.AssertAcquired();
   port->lock.AssertAcquired();
 
   for (;;) {
     ScopedMessage message;
     port->message_queue.GetNextMessageIf(nullptr, &message);
     if (!message)
       break;
 
@@ skipping 68 matching lines @@
 
   if (num_data_bytes)
     memcpy(header + 1, data, num_data_bytes);
 
   return message;
 }
 
 }  // namespace ports
 }  // namespace edk
 }  // namespace mojo