Chromium Code Reviews| Index: mojo/edk/system/message_for_transit.cc |
| diff --git a/mojo/edk/system/message_for_transit.cc b/mojo/edk/system/message_for_transit.cc |
| new file mode 100644 |
| index 0000000000000000000000000000000000000000..279126a1a80c3c31f4ed1ac6e49514bd6d99edac |
| --- /dev/null |
| +++ b/mojo/edk/system/message_for_transit.cc |
| @@ -0,0 +1,188 @@ |
| +// 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, t
Ken Rockot(use gerrit already)
2016/04/21 16:47:40
Yes, this was all broken and confusing because I w
|
| + 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 |