[mojo] Implement pipe fusion API

mojo/edk/system/message_pipe_dispatcher.cc

 // Copyright 2015 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/message_pipe_dispatcher.h"
 
 #include <limits>
 
 #include "base/macros.h"
 #include "base/memory/ref_counted.h"
@@ skipping 81 matching lines @@
       pipe_id_(pipe_id),
       endpoint_(endpoint) {
   DVLOG(2) << "Creating new MessagePipeDispatcher for port " << port.name()
            << " [pipe_id=" << pipe_id << "; endpoint=" << endpoint << "]";
 
   node_controller_->SetPortObserver(
       port_,
       make_scoped_refptr(new PortObserverThunk(this)));
 }
 
+bool MessagePipeDispatcher::BeginFuse() {
+  base::AutoLock lock(signal_lock_);
+  if (port_closed_ || in_transit_ || is_fusing_)
+    return false;
+  is_fusing_ = true;
+  return true;
+}
+
+void MessagePipeDispatcher::CancelFuse() {
+  base::AutoLock lock(signal_lock_);
+  CHECK(is_fusing_);
+  is_fusing_ = false;
+}
+
+bool MessagePipeDispatcher::CompleteFuse(MessagePipeDispatcher* other) {
+  node_controller_->SetPortObserver(port_, nullptr);
+  node_controller_->SetPortObserver(other->port_, nullptr);
+
+  ports::PortRef port0;
+  {
+    base::AutoLock lock(signal_lock_);
+    CHECK(is_fusing_);
+    port0 = port_;
+    port_closed_ = true;
+    awakables_.CancelAll();
+  }
+
+  ports::PortRef port1;
+  {
+    base::AutoLock lock(other->signal_lock_);
+    CHECK(other->is_fusing_);
+    port1 = other->port_;
+    other->port_closed_ = true;
+    other->awakables_.CancelAll();
+  }
+
+  // Both ports are always closed by this call.
+  int rv = node_controller_->MergeLocalPorts(port0, port1);
+  return rv == ports::OK;
+}
+
 Dispatcher::Type MessagePipeDispatcher::GetType() const {
   return Type::MESSAGE_PIPE;
 }
 
 MojoResult MessagePipeDispatcher::Close() {
   base::AutoLock lock(signal_lock_);
   DVLOG(1) << "Closing message pipe " << pipe_id_ << " endpoint " << endpoint_
            << " [port=" << port_.name() << "]";
   return CloseNoLock();
 }
 
 MojoResult MessagePipeDispatcher::Watch(MojoHandleSignals signals,
                                         const Watcher::WatchCallback& callback,
                                         uintptr_t context) {
   base::AutoLock lock(signal_lock_);
 
-  if (port_closed_ || in_transit_)
+  if (port_closed_ || in_transit_ || is_fusing_)
     return MOJO_RESULT_INVALID_ARGUMENT;
 
   return awakables_.AddWatcher(
       signals, callback, context, GetHandleSignalsStateNoLock());
 }
 
 MojoResult MessagePipeDispatcher::CancelWatch(uintptr_t context) {
   base::AutoLock lock(signal_lock_);
 
-  if (port_closed_ || in_transit_)
+  if (port_closed_ || in_transit_ || is_fusing_)
     return MOJO_RESULT_INVALID_ARGUMENT;
 
   return awakables_.RemoveWatcher(context);
 }
 
 MojoResult MessagePipeDispatcher::WriteMessage(
     const void* bytes,
     uint32_t num_bytes,
     const DispatcherInTransit* dispatchers,
     uint32_t num_dispatchers,
     MojoWriteMessageFlags flags) {
 
   {
     base::AutoLock lock(signal_lock_);
-    if (port_closed_ || in_transit_)
+    if (port_closed_ || in_transit_ || is_fusing_)
       return MOJO_RESULT_INVALID_ARGUMENT;
   }
 
   // A structure for retaining information about every Dispatcher we're about
   // to send. This information is collected by calling StartSerialize() on
   // each dispatcher in sequence.
   struct DispatcherInfo {
     uint32_t num_bytes;
     uint32_t num_ports;
     uint32_t num_handles;
@@ skipping 133 matching lines @@
 }
 
 MojoResult MessagePipeDispatcher::ReadMessage(void* bytes,
                                               uint32_t* num_bytes,
                                               MojoHandle* handles,
                                               uint32_t* num_handles,
                                               MojoReadMessageFlags flags) {
   {
     base::AutoLock lock(signal_lock_);
     // We can't read from a port that's closed or in transit!
-    if (port_closed_ || in_transit_)
+    if (port_closed_ || in_transit_ || is_fusing_)
       return MOJO_RESULT_INVALID_ARGUMENT;
   }
 
   bool no_space = false;
   bool may_discard = flags & MOJO_READ_MESSAGE_FLAG_MAY_DISCARD;
 
   // Ensure the provided buffers are large enough to hold the next message.
   // GetMessageIf provides an atomic way to test the next message without
   // committing to removing it from the port's underlying message queue until
   // we are sure we can consume it.
@@ skipping 134 matching lines @@
   return GetHandleSignalsStateNoLock();
 }
 
 MojoResult MessagePipeDispatcher::AddAwakable(
     Awakable* awakable,
     MojoHandleSignals signals,
     uintptr_t context,
     HandleSignalsState* signals_state) {
   base::AutoLock lock(signal_lock_);
 
-  if (port_closed_ || in_transit_) {
+  if (port_closed_ || in_transit_ || is_fusing_) {
     if (signals_state)
       *signals_state = HandleSignalsState();
     return MOJO_RESULT_INVALID_ARGUMENT;
   }
 
   HandleSignalsState state = GetHandleSignalsStateNoLock();
 
   DVLOG(2) << "Getting signal state for pipe " << pipe_id_ << " endpoint "
            << endpoint_ << " [awakable=" << awakable << "; port="
            << port_.name() << "; signals=" << signals << "; satisfied="
@@ skipping 17 matching lines @@
            << endpoint_ << " [awakable=" << awakable << "; port="
            << port_.name() << "; signals=" << signals << "]";
 
   awakables_.Add(awakable, signals, context);
   return MOJO_RESULT_OK;
 }
 
 void MessagePipeDispatcher::RemoveAwakable(Awakable* awakable,
                                            HandleSignalsState* signals_state) {
   base::AutoLock lock(signal_lock_);
-  if (port_closed_ || in_transit_) {
+  if (port_closed_ || in_transit_ || is_fusing_) {
     if (signals_state)
       *signals_state = HandleSignalsState();
   } else if (signals_state) {
     *signals_state = GetHandleSignalsStateNoLock();
   }
 
   DVLOG(2) << "Removing awakable from pipe " << pipe_id_ << " endpoint "
            << endpoint_ << " [awakable=" << awakable << "; port="
            << port_.name() << "]";
 
@@ skipping 14 matching lines @@
   SerializedState* state = static_cast<SerializedState*>(destination);
   state->pipe_id = pipe_id_;
   state->endpoint = static_cast<int8_t>(endpoint_);
   memset(state->padding, 0, sizeof(state->padding));
   ports[0] = port_.name();
   return true;
 }
 
 bool MessagePipeDispatcher::BeginTransit() {
   base::AutoLock lock(signal_lock_);
-  if (in_transit_ || port_closed_)
+  if (in_transit_ || port_closed_ || is_fusing_)
     return false;
   in_transit_ = true;
   return in_transit_;
 }
 
 void MessagePipeDispatcher::CompleteTransitAndClose() {
   node_controller_->SetPortObserver(port_, nullptr);
 
   base::AutoLock lock(signal_lock_);
   in_transit_ = false;
@@ skipping 30 matching lines @@
   return new MessagePipeDispatcher(internal::g_core->GetNodeController(), port,
                                    state->pipe_id, state->endpoint);
 }
 
 MessagePipeDispatcher::~MessagePipeDispatcher() {
   DCHECK(port_closed_ && !in_transit_);
 }
 
 MojoResult MessagePipeDispatcher::CloseNoLock() {
   signal_lock_.AssertAcquired();
-  if (port_closed_ || in_transit_)
+  if (port_closed_ || in_transit_ || is_fusing_)
     return MOJO_RESULT_INVALID_ARGUMENT;
 
   port_closed_ = true;
   awakables_.CancelAll();
 
   if (!port_transferred_) {
     base::AutoUnlock unlock(signal_lock_);
     node_controller_->ClosePort(port_);
   }
 
@@ skipping 56 matching lines @@
     DVLOG(1) << "Peer closure detected on message pipe " << pipe_id_
              << " endpoint " << endpoint_ << " [port=" << port_.name() << "]";
   }
 #endif
 
   awakables_.AwakeForStateChange(GetHandleSignalsStateNoLock());
 }
 
 }  // namespace edk
 }  // namespace mojo