[mojo] Implement pipe fusion API

mojo/edk/system/core.cc

 // Copyright 2013 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/core.h"
 
 #include <string.h>
 
 #include <utility>
 
@@ skipping 433 matching lines @@
   *message_pipe_handle0 = AddDispatcher(
       new MessagePipeDispatcher(GetNodeController(), port0, pipe_id, 0));
   if (*message_pipe_handle0 == MOJO_HANDLE_INVALID)
     return MOJO_RESULT_RESOURCE_EXHAUSTED;
 
   *message_pipe_handle1 = AddDispatcher(
       new MessagePipeDispatcher(GetNodeController(), port1, pipe_id, 1));
   if (*message_pipe_handle1 == MOJO_HANDLE_INVALID) {
     scoped_refptr<Dispatcher> unused;
     unused->Close();
+
+    base::AutoLock lock(handles_lock_);
     handles_.GetAndRemoveDispatcher(*message_pipe_handle0, &unused);
     return MOJO_RESULT_RESOURCE_EXHAUSTED;
   }
 
   return MOJO_RESULT_OK;
 }
 
 MojoResult Core::WriteMessage(MojoHandle message_pipe_handle,
                               const void* bytes,
                               uint32_t num_bytes,
@@ skipping 52 matching lines @@
                              MojoReadMessageFlags flags) {
   RequestContext request_context;
   CHECK((!num_handles || !*num_handles || handles) &&
         (!num_bytes || !*num_bytes || bytes));
   auto dispatcher = GetDispatcher(message_pipe_handle);
   if (!dispatcher)
     return MOJO_RESULT_INVALID_ARGUMENT;
   return dispatcher->ReadMessage(bytes, num_bytes, handles, num_handles, flags);
 }
 
+MojoResult Core::FuseMessagePipes(MojoHandle handle0, MojoHandle handle1) {
+  RequestContext request_context;
+  scoped_refptr<Dispatcher> dispatcher0 = GetDispatcher(handle0);
+  scoped_refptr<Dispatcher> dispatcher1 = GetDispatcher(handle1);
+
+  if (!dispatcher0 || !dispatcher1)
+    return MOJO_RESULT_INVALID_ARGUMENT;
+
+  if (dispatcher0->GetType() != Dispatcher::Type::MESSAGE_PIPE ||
+      dispatcher1->GetType() != Dispatcher::Type::MESSAGE_PIPE) {
+    return MOJO_RESULT_INVALID_ARGUMENT;
+  }
+
+  // Both handles are valid message pipe handles. Beyond this point, both
+  // handles are always invalidated.
+  {
+    base::AutoLock lock(handles_lock_);
+
+    // Someone may have closed the handles while we were checking them above, so
+    // we need to handle that as we remove them from the handle table.
+    MojoResult result0 = handles_.GetAndRemoveDispatcher(handle0, &dispatcher0);
+    MojoResult result1 = handles_.GetAndRemoveDispatcher(handle1, &dispatcher1);
+    if (result0 != MOJO_RESULT_OK)
+      dispatcher0 = nullptr;
+    if (result1 != MOJO_RESULT_OK)
+      dispatcher1 = nullptr;
+
+    if (!dispatcher0 || !dispatcher1) {
+      if (dispatcher0)
+        dispatcher0->Close();
+      if (dispatcher1)
+        dispatcher1->Close();
+      return MOJO_RESULT_FAILED_PRECONDITION;
+    }
+  }
+
+  DCHECK(dispatcher0 && dispatcher1);
+
+  MessagePipeDispatcher* mpd0 =
+      static_cast<MessagePipeDispatcher*>(dispatcher0.get());
+  MessagePipeDispatcher* mpd1 =
+      static_cast<MessagePipeDispatcher*>(dispatcher1.get());
+
+  if (!mpd0->BeginFuse()) {
+    mpd0->Close();
+    mpd1->Close();
+    return MOJO_RESULT_FAILED_PRECONDITION;
+  }
+
+  if (!mpd1->BeginFuse()) {
+    mpd0->CancelFuse();
+    mpd0->Close();
+    mpd1->Close();
+    return MOJO_RESULT_FAILED_PRECONDITION;
+  }
+
+  if (!mpd0->CompleteFuse(mpd1))
+    return MOJO_RESULT_FAILED_PRECONDITION;
+
+  return MOJO_RESULT_OK;
+}
+
 MojoResult Core::CreateDataPipe(
     const MojoCreateDataPipeOptions* options,
     MojoHandle* data_pipe_producer_handle,
     MojoHandle* data_pipe_consumer_handle) {
   RequestContext request_context;
   if (options && options->struct_size != sizeof(MojoCreateDataPipeOptions))
     return MOJO_RESULT_INVALID_ARGUMENT;
 
   MojoCreateDataPipeOptions create_options;
   create_options.struct_size = sizeof(MojoCreateDataPipeOptions);
@@ skipping 25 matching lines @@
   scoped_refptr<Dispatcher> consumer = new DataPipeConsumerDispatcher(
       GetNodeController(), port1, ring_buffer, create_options,
       true /* initialized */, pipe_id);
 
   *data_pipe_producer_handle = AddDispatcher(producer);
   *data_pipe_consumer_handle = AddDispatcher(consumer);
   if (*data_pipe_producer_handle == MOJO_HANDLE_INVALID ||
       *data_pipe_consumer_handle == MOJO_HANDLE_INVALID) {
     if (*data_pipe_producer_handle != MOJO_HANDLE_INVALID) {
       scoped_refptr<Dispatcher> unused;
+      base::AutoLock lock(handles_lock_);
       handles_.GetAndRemoveDispatcher(*data_pipe_producer_handle, &unused);
     }
     producer->Close();
     consumer->Close();
     return MOJO_RESULT_RESOURCE_EXHAUSTED;
   }
 
   return MOJO_RESULT_OK;
 }
 
@@ skipping 238 matching lines @@
     scoped_ptr<NodeController> node_controller) {
   // It's OK to leak this reference. At this point we know the IO loop is still
   // running, and we know the NodeController will observe its eventual
   // destruction. This tells the NodeController to delete itself when that
   // happens.
   node_controller.release()->DestroyOnIOThreadShutdown();
 }
 
 }  // namespace edk
 }  // namespace mojo