[mojo] Add CreateSharedBuffer method to Broker.

mojo/edk/system/child_broker.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/child_broker.h"
 
 #include <stddef.h>
 #include <stdint.h>
 
 #include <utility>
 
 #include "base/bind.h"
 #include "base/logging.h"
 #include "mojo/edk/embedder/embedder_internal.h"
 #include "mojo/edk/embedder/platform_channel_pair.h"
+#include "mojo/edk/embedder/platform_shared_buffer.h"
+#include "mojo/edk/embedder/platform_support.h"
 #include "mojo/edk/system/broker_messages.h"
 #include "mojo/edk/system/message_pipe_dispatcher.h"
 
+#if defined(OS_POSIX)
+#include "mojo/edk/embedder/platform_channel_utils_posix.h"
+#endif
+
 namespace mojo {
 namespace edk {
 
 ChildBroker* ChildBroker::GetInstance() {
   return base::Singleton<
       ChildBroker, base::LeakySingletonTraits<ChildBroker>>::get();
 }
 
 void ChildBroker::SetChildBrokerHostHandle(ScopedPlatformHandle handle)  {
   ScopedPlatformHandle parent_async_channel_handle;
+  parent_sync_channel_ = std::move(handle);
+
+// We have two pipes to the parent. The first is for the token
+// exchange for creating and passing handles on Windows, and creating shared
+// buffers on POSIX, since the child needs the parent's help if it is
+// sandboxed. The second is used for multiplexing related messages. We
+// send the second over the first.
 #if defined(OS_POSIX)
-  parent_async_channel_handle = std::move(handle);
+  std::deque<PlatformHandle> received_handles;
+  char buf[1];
+  ssize_t result = PlatformChannelRecvmsg(parent_sync_channel_.get(), buf, 1,

[Eliot Courtney, 2016/01/05 04:45:47]

Does this need to be queried in a loop since it won't block? It seemed to me
that the Windows version of this code (ReadFile...) doesn't block in this case,
so I think it might have the same problem if this is a problem.

[jam, 2016/01/06 02:58:16]

ReadFile below specifies the number of bytes to read ("sizeof(parent_handle)")
and no overlapped struct (the NULL parameter), so it's synchronous.

I'm not an expert on posix by any means, but can't you simulate the above by
temporarily calling
fcntl(parent_sync_channel_.get(), F_SETFL, 0)
on the pipe?

[Eliot Courtney, 2016/01/07 02:26:09]

Thanks! I've done this, although made it not temporary, since it seems
reasonable parent_sync_channel_ should always be synchronous.

[jam, 2016/01/08 22:37:40]

great, i had some more statements in this comment about why it's temporary but
then i removed them (i agree it should always be sync). i just forgot this one
place :)

+                                          &received_handles);
+  CHECK_EQ(1, result);
+  CHECK_EQ(1u, received_handles.size());
+  parent_async_channel_handle.reset(received_handles.front());
 #else
-  // On Windows we have two pipes to the parent. The first is for the token
-  // exchange for creating and passing handles, since the child needs the
-  // parent's help if it is sandboxed. The second is the same as POSIX, which is
-  // used for multiplexing related messages. So on Windows, we send the second
-  // pipe as the first string over the first one.
-  parent_sync_channel_ = handle.Pass();
-
   HANDLE parent_handle = INVALID_HANDLE_VALUE;
   DWORD bytes_read = 0;
   BOOL rv = ReadFile(parent_sync_channel_.get().handle, &parent_handle,
                      sizeof(parent_handle), &bytes_read, NULL);
   CHECK(rv);
   parent_async_channel_handle.reset(PlatformHandle(parent_handle));
+#endif
   sync_channel_lock_.Unlock();
-#endif
 
   internal::g_io_thread_task_runner->PostTask(
       FROM_HERE,
       base::Bind(&ChildBroker::InitAsyncChannel, base::Unretained(this),
                  base::Passed(&parent_async_channel_handle)));
 }
 
 #if defined(OS_WIN)
 void ChildBroker::CreatePlatformChannelPair(
     ScopedPlatformHandle* server, ScopedPlatformHandle* client) {
@@ skipping 35 matching lines @@
   std::vector<HANDLE> handles_temp(count);
   uint32_t response_size =
       static_cast<uint32_t>(handles_temp.size()) * sizeof(HANDLE);
   sync_channel_lock_.Lock();
   if (WriteAndReadResponse(message, &handles_temp[0], response_size)) {
     for (uint32_t i = 0; i < count; ++i)
       handles[i].handle = handles_temp[i];
     sync_channel_lock_.Unlock();
   }
 }
+#else
+scoped_refptr<PlatformSharedBuffer> ChildBroker::CreateSharedBuffer(
+    size_t num_bytes) {
+  sync_channel_lock_.Lock();
+  scoped_refptr<PlatformSharedBuffer> shared_buffer =
+      CreateSharedBufferNoLock(num_bytes);

[jam, 2016/01/06 02:58:16]

no need to have two methods for this. you can inline that second method here.
CreatePlatformChannelPair is an exception because it is called initially with
the lock already acquired.

[Eliot Courtney, 2016/01/07 02:26:09]

Done.

+  sync_channel_lock_.Unlock();
+
+  return shared_buffer;
+}
 #endif
 
 void ChildBroker::ConnectMessagePipe(uint64_t pipe_id,
                                      MessagePipeDispatcher* message_pipe) {
   DCHECK(internal::g_io_thread_task_runner->RunsTasksOnCurrentThread());
 
   ConnectMessagePipeMessage data;
   memset(&data, 0, sizeof(data));
   data.pipe_id = pipe_id;
   if (pending_connects_.find(pipe_id) != pending_connects_.end()) {
@@ skipping 53 matching lines @@
   connected_pipes_[message_pipe]->RemoveRoute(pipe_id);
   connected_pipes_.erase(message_pipe);
 }
 
 ChildBroker::ChildBroker()
     : parent_async_channel_(nullptr),
       in_process_pipes_channel1_(nullptr),
       in_process_pipes_channel2_(nullptr) {
   DCHECK(!internal::g_broker);
   internal::g_broker = this;
-#if defined(OS_WIN)
   // Block any threads from calling this until we have a pipe to the parent.
   sync_channel_lock_.Lock();
-#endif
 }
 
 ChildBroker::~ChildBroker() {
 }
 
 void ChildBroker::OnReadMessage(
     const MessageInTransit::View& message_view,
     ScopedPlatformHandleVectorPtr platform_handles) {
   DCHECK(internal::g_io_thread_task_runner->RunsTasksOnCurrentThread());
   MultiplexMessages type =
@@ skipping 89 matching lines @@
   // then when it's read it returns no messages because it doesn't have the
   // channel yet.
   message_pipe->GotNonTransferableChannel(raw_channel->channel());
   // The above call could have caused |CloseMessagePipe| to be called.
   if (connected_pipes_.find(message_pipe) != connected_pipes_.end())
     raw_channel->AddRoute(pipe_id, message_pipe);
 }
 
 #if defined(OS_WIN)
 
+void ChildBroker::CreatePlatformChannelPairNoLock(
+    ScopedPlatformHandle* server,
+    ScopedPlatformHandle* client) {
+  BrokerMessage message;
+  message.size = kBrokerMessageHeaderSize;
+  message.id = CREATE_PLATFORM_CHANNEL_PAIR;
+
+  uint32_t response_size = 2 * sizeof(HANDLE);
+  HANDLE handles[2];
+  if (WriteAndReadResponse(&message, handles, response_size)) {
+    server->reset(PlatformHandle(handles[0]));
+    client->reset(PlatformHandle(handles[1]));
+  }
+}
+
 bool ChildBroker::WriteAndReadResponse(BrokerMessage* message,
                                        void* response,
                                        uint32_t response_size) {
   CHECK(parent_sync_channel_.is_valid());
 
   bool result = true;
+

[jam, 2016/01/06 02:58:16]

nit: no need to change

[Eliot Courtney, 2016/01/07 02:26:09]

Done.

   DWORD bytes_written = 0;
   // This will always write in one chunk per
   // https://msdn.microsoft.com/en-us/library/windows/desktop/aa365150.aspx.
   BOOL rv = WriteFile(parent_sync_channel_.get().handle, message, message->size,
                       &bytes_written, NULL);
   if (!rv || bytes_written != message->size) {
     LOG(ERROR) << "Child token serializer couldn't write message.";
     result = false;
   } else {
     while (response_size) {
       DWORD bytes_read = 0;
       rv = ReadFile(parent_sync_channel_.get().handle, response, response_size,
                     &bytes_read, NULL);
       if (!rv) {
         LOG(ERROR) << "Child token serializer couldn't read result.";
         result = false;
         break;
       }
       response_size -= bytes_read;
       response = static_cast<char*>(response) + bytes_read;
     }
   }
 
   return result;
 }
+#else
+scoped_refptr<PlatformSharedBuffer> ChildBroker::CreateSharedBufferNoLock(
+    size_t num_bytes) {
+  BrokerMessage message;
+  message.size = kBrokerMessageHeaderSize + sizeof(uint32_t);
+  message.id = CREATE_SHARED_BUFFER;
+  message.num_bytes = num_bytes;
 
-void ChildBroker::CreatePlatformChannelPairNoLock(
-    ScopedPlatformHandle* server, ScopedPlatformHandle* client) {
-  BrokerMessage message;
-  message.size = kBrokerMessageHeaderSize;
-  message.id = CREATE_PLATFORM_CHANNEL_PAIR;
+  std::deque<PlatformHandle> handles;
+  if (WriteAndReadHandles(&message, &handles)) {
+    DCHECK_EQ(1u, handles.size());
+    if (handles.empty()) {
+      LOG(ERROR)
+          << "ChildBroker did not receive handle when creating shared buffer";
+      return nullptr;
+    }
 
-  uint32_t response_size = 2 * sizeof(HANDLE);
-  HANDLE handles[2];
-  if (WriteAndReadResponse(&message, handles, response_size)) {
-    server->reset(PlatformHandle(handles[0]));
-    client->reset(PlatformHandle(handles[1]));
+    PlatformHandle handle = handles.front();
+    if (handle.is_valid())
+      return internal::g_platform_support->CreateSharedBufferFromHandle(
+          num_bytes, ScopedPlatformHandle(handles.front()));
+    else
+      return nullptr;
   }
+
+  return nullptr;
 }
 
+bool ChildBroker::WriteAndReadHandles(BrokerMessage* message,
+                                      std::deque<PlatformHandle>* handles) {
+  CHECK(parent_sync_channel_.is_valid());
+
+  uint32_t remaining_bytes = message->size;
+  while (remaining_bytes > 0) {
+    ssize_t bytes_written = PlatformChannelWrite(

[Eliot Courtney, 2016/01/05 04:45:47]

This loop is on top of a non-blocking function (PlatformChannelWrite), which
could eat CPU; Is it worth somehow making this block? 

[jam, 2016/01/06 02:58:16]

i don't see how it would ever block in practice, because we only write tiny
messages one at a time?

[Eliot Courtney, 2016/01/07 02:26:09]

I removed the O_NONBLOCK using fcntl like you suggested, so I think my concern
no longer applies.

+        parent_sync_channel_.get(),
+        reinterpret_cast<uint8_t*>(message) + (message->size - remaining_bytes),
+        remaining_bytes);
+
+    if (bytes_written == -1 && errno != EAGAIN && errno != EWOULDBLOCK)
+      return false;
+
+    if (bytes_written != -1)
+      remaining_bytes -= bytes_written;
+  }
+
+  while (1) {

[jam, 2016/01/06 02:58:16]

same as first comment in this file, why not make reads on this pipe always be
blocking?

[Eliot Courtney, 2016/01/07 02:26:09]

Done.

+    char buf[1];
+    ssize_t bytes_read =
+        PlatformChannelRecvmsg(parent_sync_channel_.get(), buf, 1, handles);

[Eliot Courtney, 2016/01/05 04:45:47]

Same comment as above.

[Eliot Courtney, 2016/01/07 02:26:08]

Done.

+    if (bytes_read == 0 ||
+        (bytes_read == -1 && errno != EAGAIN && errno != EWOULDBLOCK))
+      return false;
+
+    if (handles->size() >= 1)
+      break;
+  }
+  return true;
+}
 #endif
 
 }  // namespace edk
 }  // namespace mojo