[mojo] Implement pipe fusion API

mojo/edk/system/core.cc

Index: mojo/edk/system/core.cc
diff --git a/mojo/edk/system/core.cc b/mojo/edk/system/core.cc
index a54e8b10f446f7645f103c201a4bd986d68a57b3..dd5e43ead9cac188614e2921f854258f1d2b8fd7 100644
--- a/mojo/edk/system/core.cc
+++ b/mojo/edk/system/core.cc
@@ -451,6 +451,8 @@ MojoResult Core::CreateMessagePipe(
   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;
   }
@@ -523,6 +525,68 @@ MojoResult Core::ReadMessage(MojoHandle message_pipe_handle,
   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,
@@ -568,6 +632,7 @@ MojoResult Core::CreateDataPipe(
       *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();