[mojo-edk] Add explicit message object APIs

mojo/edk/system/message_for_transit.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/message_for_transit.h"
+
+#include <vector>
+
+#include "mojo/edk/embedder/platform_handle_vector.h"
+
+namespace mojo {
+namespace edk {
+
+namespace {
+
+static_assert(sizeof(MessageForTransit::MessageHeader) % 8 == 0,
+              "Invalid MessageHeader size.");
+static_assert(sizeof(MessageForTransit::DispatcherHeader) % 8 == 0,
+              "Invalid DispatcherHeader size.");
+
+const MessageForTransit::MessageHeader* GetHeader(const PortsMessage& message) {
+  return static_cast<const MessageForTransit::MessageHeader*>(
+      message.payload_bytes());
+}
+
+}  // namespace
+
+MessageForTransit::~MessageForTransit() {
+  if (message_) {
+    // Release all the platform handles. Note that because |message_| is still
+    // valid here, we know the message was not sent. In that case, the local
+    // dispatchers still retain ownership of these handles. Releasing them here
+    // is not a leak.

[Anand Mistry (off Chromium), 2016/04/21 12:19:42]

I don't believe this logic is correct. Actually, the whole ownership between
this, WriteMessage, and WriteMessageNew is confusing me.

Consider the case where you're using CreateMessage with a
PlatformHandleDispatcher. When the message creation succeeds,
PlatformHandleDispatcher::CompleteTransitAndClose() runs and releases it's
ownership of the PlatformHandle, effectively transferring ownership to this
message. Therefore, if the message send fails, the PlatformHandle leaks because
this doesn't close it.

However, if you look at WriteMessage, it only calls CompleteTransitAndClose() if
the write succeeds.

Now, where it's more confusing is that this message DOES take ownership of
handles, but then releases it if nothing takes over message_. But also, if the
MessagePipeDispatcher::WriteMessage fails after taking over message_, ownership
of handles reverts to the PlatformHandleDispatcher via CancelTransit. But
PortsMessage also owns its handles resulting in a double-free.

[Ken Rockot(use gerrit already), 2016/04/21 16:47:40]

Yes, this was all broken and confusing because I was trying to maintain the
original behavior of WriteMessage while implementing it in terms of message
objects. This was really only to preserve behavior expected by tests though
(namely that a failed WriteMessage leaves sent handles open), so I've decided to
change that behavior and update the test expectations. Bindings don't rely on
being able to recover from WriteMessage failure so it seems like a pointless
behavior to preserve.

WriteMessage is simpler now and the lifetime of handles should be clear.

+    Channel::MessagePtr m = message_->TakeChannelMessage();
+    ScopedPlatformHandleVectorPtr handles = m->TakeHandles();
+    if (handles)
+      handles->clear();
+  }
+}
+
+// static
+MojoResult MessageForTransit::Create(
+    MessageForTransit** message,
+    uint32_t num_bytes,
+    const Dispatcher::DispatcherInTransit* dispatchers,
+    uint32_t num_dispatchers) {
+  // A structure for retaining information about every Dispatcher that will be
+  // sent with this message.
+  struct DispatcherInfo {
+    uint32_t num_bytes;
+    uint32_t num_ports;
+    uint32_t num_handles;
+  };
+
+  // This is only the base header size. It will grow as we accumulate the
+  // size of serialized state for each dispatcher.
+  size_t header_size = sizeof(MessageHeader) +
+      num_dispatchers * sizeof(DispatcherHeader);
+  size_t num_ports = 0;
+  size_t num_handles = 0;
+
+  std::vector<DispatcherInfo> dispatcher_info(num_dispatchers);
+  for (size_t i = 0; i < num_dispatchers; ++i) {
+    Dispatcher* d = dispatchers[i].dispatcher.get();
+    d->StartSerialize(&dispatcher_info[i].num_bytes,
+                      &dispatcher_info[i].num_ports,
+                      &dispatcher_info[i].num_handles);
+    header_size += dispatcher_info[i].num_bytes;
+    num_ports += dispatcher_info[i].num_ports;
+    num_handles += dispatcher_info[i].num_handles;
+  }
+
+  // We now have enough information to fully allocate the message storage.
+  std::unique_ptr<PortsMessage> msg = PortsMessage::NewUserMessage(
+      header_size + num_bytes, num_ports, num_handles);
+  if (!msg)
+    return MOJO_RESULT_RESOURCE_EXHAUSTED;
+
+  // Populate the message header with information about serialized dispatchers.
+  //
+  // The front of the message is always a MessageHeader followed by a
+  // DispatcherHeader for each dispatcher to be sent.
+  MessageHeader* header =
+      static_cast<MessageHeader*>(msg->mutable_payload_bytes());
+  DispatcherHeader* dispatcher_headers =
+      reinterpret_cast<DispatcherHeader*>(header + 1);
+
+  // Serialized dispatcher state immediately follows the series of
+  // DispatcherHeaders.
+  char* dispatcher_data =
+      reinterpret_cast<char*>(dispatcher_headers + num_dispatchers);
+
+  header->num_dispatchers = num_dispatchers;
+
+  // |header_size| is the total number of bytes preceding the message payload,
+  // including all dispatcher headers and serialized dispatcher state.
+  DCHECK_LE(header_size, std::numeric_limits<uint32_t>::max());
+  header->header_size = static_cast<uint32_t>(header_size);
+
+  if (num_dispatchers > 0) {
+    ScopedPlatformHandleVectorPtr handles(
+        new PlatformHandleVector(num_handles));
+    size_t port_index = 0;
+    size_t handle_index = 0;
+    bool fail = false;
+    for (size_t i = 0; i < num_dispatchers; ++i) {
+      Dispatcher* d = dispatchers[i].dispatcher.get();
+      DispatcherHeader* dh = &dispatcher_headers[i];
+      const DispatcherInfo& info = dispatcher_info[i];
+
+      // Fill in the header for this dispatcher.
+      dh->type = static_cast<int32_t>(d->GetType());
+      dh->num_bytes = info.num_bytes;
+      dh->num_ports = info.num_ports;
+      dh->num_platform_handles = info.num_handles;
+
+      // Fill in serialized state, ports, and platform handles. We'll cancel
+      // the send if the dispatcher implementation rejects for some reason.
+      if (!d->EndSerialize(static_cast<void*>(dispatcher_data),
+                           msg->mutable_ports() + port_index,
+                           handles->data() + handle_index)) {
+        fail = true;
+        break;
+      }
+
+      dispatcher_data += info.num_bytes;
+      port_index += info.num_ports;
+      handle_index += info.num_handles;
+    }
+
+    if (fail) {
+      // Release any platform handles we've accumulated. Their dispatchers
+      // retain ownership when message creation fails, so these are not actually
+      // leaking.
+      handles->clear();
+      return MOJO_RESULT_INVALID_ARGUMENT;
+    }
+
+    // Take ownership of all the handles and move them into message storage.
+    msg->SetHandles(std::move(handles));
+  }
+
+  *message = new MessageForTransit(std::move(msg));
+  return MOJO_RESULT_OK;
+}
+
+// static
+MessageForTransit* MessageForTransit::WrapPortsMessage(
+    std::unique_ptr<PortsMessage> message) {
+  return new MessageForTransit(std::move(message));
+}
+
+const void* MessageForTransit::bytes() const {
+  DCHECK(message_);
+  return static_cast<const void*>(
+      static_cast<const char*>(message_->payload_bytes()) +
+          GetHeader(*message_)->header_size);
+}
+
+void* MessageForTransit::mutable_bytes() {
+  DCHECK(message_);
+  return static_cast<void*>(
+      static_cast<char*>(message_->mutable_payload_bytes()) +
+          GetHeader(*message_)->header_size);
+}
+
+size_t MessageForTransit::num_bytes() const {
+  DCHECK(message_);
+  size_t header_size = GetHeader(*message_)->header_size;
+  DCHECK_GE(message_->num_payload_bytes(), header_size);
+  return message_->num_payload_bytes() - header_size;
+}
+
+size_t MessageForTransit::num_handles() const {
+  DCHECK(message_);
+  return GetHeader(*message_)->num_dispatchers;
+}
+
+std::unique_ptr<PortsMessage> MessageForTransit::TakePortsMessage() {
+  return std::move(message_);
+}
+
+MessageForTransit::MessageForTransit(std::unique_ptr<PortsMessage> message)
+    : message_(std::move(message)) {
+}
+
+}  // namespace edk
+}  // namespace mojo