Update mojo sdk to rev f68e697e389943cd9bf9652397312280e96b127a

third_party/mojo/src/mojo/edk/system/remote_consumer_data_pipe_impl.cc

Index: third_party/mojo/src/mojo/edk/system/remote_consumer_data_pipe_impl.cc
diff --git a/third_party/mojo/src/mojo/edk/system/remote_consumer_data_pipe_impl.cc b/third_party/mojo/src/mojo/edk/system/remote_consumer_data_pipe_impl.cc
new file mode 100644
index 0000000000000000000000000000000000000000..fa59df06e6ecc830582a7698ff7e9b4ff009c904
--- /dev/null
+++ b/third_party/mojo/src/mojo/edk/system/remote_consumer_data_pipe_impl.cc
@@ -0,0 +1,418 @@
+// 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/remote_consumer_data_pipe_impl.h"
+
+#include <string.h>
+
+#include <algorithm>
+
+#include "base/logging.h"
+#include "base/memory/scoped_ptr.h"
+#include "mojo/edk/system/channel.h"
+#include "mojo/edk/system/channel_endpoint.h"
+#include "mojo/edk/system/configuration.h"
+#include "mojo/edk/system/data_pipe.h"
+#include "mojo/edk/system/message_in_transit.h"
+#include "mojo/edk/system/remote_data_pipe_ack.h"
+
+namespace mojo {
+namespace system {
+
+namespace {
+
+bool ValidateIncomingMessage(size_t element_num_bytes,
+                             size_t capacity_num_bytes,
+                             size_t consumer_num_bytes,
+                             const MessageInTransit* message) {
+  // We should only receive endpoint messages.
+  DCHECK_EQ(message->type(), MessageInTransit::kTypeEndpoint);
+
+  // But we should check the subtype; only take data pipe acks.
+  if (message->subtype() != MessageInTransit::kSubtypeEndpointDataPipeAck) {
+    LOG(WARNING) << "Received message of unexpected subtype: "
+                 << message->subtype();
+    return false;
+  }
+
+  if (message->num_bytes() != sizeof(RemoteDataPipeAck)) {
+    LOG(WARNING) << "Incorrect message size: " << message->num_bytes()
+                 << " bytes (expected: " << sizeof(RemoteDataPipeAck)
+                 << " bytes)";
+    return false;
+  }
+
+  const RemoteDataPipeAck* ack =
+      static_cast<const RemoteDataPipeAck*>(message->bytes());
+  size_t num_bytes_consumed = ack->num_bytes_consumed;
+
+  if (num_bytes_consumed > consumer_num_bytes) {
+    LOG(WARNING) << "Number of bytes consumed too large: " << num_bytes_consumed
+                 << " bytes (outstanding: " << consumer_num_bytes << " bytes)";
+    return false;
+  }
+
+  if (num_bytes_consumed % element_num_bytes != 0) {
+    LOG(WARNING) << "Number of bytes consumed not a multiple of element size: "
+                 << num_bytes_consumed
+                 << " bytes (element size: " << element_num_bytes << " bytes)";
+    return false;
+  }
+
+  return true;
+}
+
+}  // namespace
+
+RemoteConsumerDataPipeImpl::RemoteConsumerDataPipeImpl(
+    ChannelEndpoint* channel_endpoint,
+    size_t consumer_num_bytes)
+    : channel_endpoint_(channel_endpoint),
+      consumer_num_bytes_(consumer_num_bytes) {
+  // Note: |buffer_| is lazily allocated.
+}
+
+RemoteConsumerDataPipeImpl::~RemoteConsumerDataPipeImpl() {
+}
+
+void RemoteConsumerDataPipeImpl::ProducerClose() {
+  if (!consumer_open()) {
+    DCHECK(!channel_endpoint_);
+    return;
+  }
+
+  Disconnect();
+}
+
+// static
+bool RemoteConsumerDataPipeImpl::ProcessMessagesFromIncomingEndpoint(
+    const MojoCreateDataPipeOptions& validated_options,
+    size_t* consumer_num_bytes,
+    MessageInTransitQueue* messages) {
+  const size_t element_num_bytes = validated_options.element_num_bytes;
+  const size_t capacity_num_bytes = validated_options.capacity_num_bytes;
+
+  if (messages) {
+    while (!messages->IsEmpty()) {
+      scoped_ptr<MessageInTransit> message(messages->GetMessage());
+      if (!ValidateIncomingMessage(element_num_bytes, capacity_num_bytes,
+                                   *consumer_num_bytes, message.get())) {
+        messages->Clear();
+        return false;
+      }
+
+      const RemoteDataPipeAck* ack =
+          static_cast<const RemoteDataPipeAck*>(message->bytes());
+      size_t num_bytes_consumed = ack->num_bytes_consumed;
+      *consumer_num_bytes -= num_bytes_consumed;
+    }
+  }
+
+  return true;
+}
+
+MojoResult RemoteConsumerDataPipeImpl::ProducerWriteData(
+    UserPointer<const void> elements,
+    UserPointer<uint32_t> num_bytes,
+    uint32_t max_num_bytes_to_write,
+    uint32_t min_num_bytes_to_write) {
+  DCHECK_EQ(max_num_bytes_to_write % element_num_bytes(), 0u);
+  DCHECK_EQ(min_num_bytes_to_write % element_num_bytes(), 0u);
+  DCHECK_GT(max_num_bytes_to_write, 0u);
+  DCHECK_GE(max_num_bytes_to_write, min_num_bytes_to_write);
+  DCHECK(consumer_open());
+  DCHECK(channel_endpoint_);
+
+  DCHECK_LE(consumer_num_bytes_, capacity_num_bytes());
+  DCHECK_EQ(consumer_num_bytes_ % element_num_bytes(), 0u);
+
+  if (min_num_bytes_to_write > capacity_num_bytes() - consumer_num_bytes_)
+    return MOJO_RESULT_OUT_OF_RANGE;
+
+  size_t num_bytes_to_write =
+      std::min(static_cast<size_t>(max_num_bytes_to_write),
+               capacity_num_bytes() - consumer_num_bytes_);
+  if (num_bytes_to_write == 0)
+    return MOJO_RESULT_SHOULD_WAIT;
+
+  // The maximum amount of data to send per message (make it a multiple of the
+  // element size.
+  // TODO(vtl): Copied from |LocalDataPipeImpl::ConvertDataToMessages()|.
+  size_t max_message_num_bytes = GetConfiguration().max_message_num_bytes;
+  max_message_num_bytes -= max_message_num_bytes % element_num_bytes();
+  DCHECK_GT(max_message_num_bytes, 0u);
+
+  size_t offset = 0;
+  while (offset < num_bytes_to_write) {
+    size_t message_num_bytes =
+        std::min(max_message_num_bytes, num_bytes_to_write - offset);
+    scoped_ptr<MessageInTransit> message(new MessageInTransit(
+        MessageInTransit::kTypeEndpoint, MessageInTransit::kSubtypeEndpointData,
+        static_cast<uint32_t>(message_num_bytes), elements.At(offset)));
+    if (!channel_endpoint_->EnqueueMessage(message.Pass())) {
+      Disconnect();
+      break;
+    }
+
+    offset += message_num_bytes;
+    consumer_num_bytes_ += message_num_bytes;
+  }
+
+  DCHECK_LE(consumer_num_bytes_, capacity_num_bytes());
+  // TODO(vtl): We report |num_bytes_to_write|, instead of |offset|, even if we
+  // failed at some point. This is consistent with the idea that writes either
+  // "succeed" or "fail" (and since some bytes may have been sent, we opt for
+  // "succeed"). Think about this some more.
+  num_bytes.Put(static_cast<uint32_t>(num_bytes_to_write));
+  return MOJO_RESULT_OK;
+}
+
+MojoResult RemoteConsumerDataPipeImpl::ProducerBeginWriteData(
+    UserPointer<void*> buffer,
+    UserPointer<uint32_t> buffer_num_bytes,
+    uint32_t min_num_bytes_to_write) {
+  DCHECK(consumer_open());
+  DCHECK(channel_endpoint_);
+
+  DCHECK_LE(consumer_num_bytes_, capacity_num_bytes());
+  DCHECK_EQ(consumer_num_bytes_ % element_num_bytes(), 0u);
+
+  size_t max_num_bytes_to_write = capacity_num_bytes() - consumer_num_bytes_;
+  if (min_num_bytes_to_write > max_num_bytes_to_write) {
+    // Don't return "should wait" since you can't wait for a specified amount
+    // of data.
+    return MOJO_RESULT_OUT_OF_RANGE;
+  }
+
+  // Don't go into a two-phase write if there's no room.
+  if (max_num_bytes_to_write == 0)
+    return MOJO_RESULT_SHOULD_WAIT;
+
+  EnsureBuffer();
+  buffer.Put(buffer_.get());
+  buffer_num_bytes.Put(static_cast<uint32_t>(max_num_bytes_to_write));
+  set_producer_two_phase_max_num_bytes_written(
+      static_cast<uint32_t>(max_num_bytes_to_write));
+  return MOJO_RESULT_OK;
+}
+
+MojoResult RemoteConsumerDataPipeImpl::ProducerEndWriteData(
+    uint32_t num_bytes_written) {
+  DCHECK_LE(num_bytes_written, producer_two_phase_max_num_bytes_written());
+  DCHECK_EQ(num_bytes_written % element_num_bytes(), 0u);
+  DCHECK_LE(num_bytes_written, capacity_num_bytes() - consumer_num_bytes_);
+
+  // TODO(vtl): The following code is copied almost verbatim from
+  // |ProducerWriteData()| (it's touchy to factor it out since it uses a
+  // |UserPointer| while we have a plain pointer.
+
+  // The maximum amount of data to send per message (make it a multiple of the
+  // element size.
+  // TODO(vtl): Copied from |LocalDataPipeImpl::ConvertDataToMessages()|.
+  size_t max_message_num_bytes = GetConfiguration().max_message_num_bytes;
+  max_message_num_bytes -= max_message_num_bytes % element_num_bytes();
+  DCHECK_GT(max_message_num_bytes, 0u);
+
+  size_t offset = 0;
+  while (offset < num_bytes_written) {
+    size_t message_num_bytes =
+        std::min(max_message_num_bytes, num_bytes_written - offset);
+    scoped_ptr<MessageInTransit> message(new MessageInTransit(
+        MessageInTransit::kTypeEndpoint, MessageInTransit::kSubtypeEndpointData,
+        static_cast<uint32_t>(message_num_bytes), buffer_.get() + offset));
+    if (!channel_endpoint_->EnqueueMessage(message.Pass())) {
+      Disconnect();
+      break;
+    }
+
+    offset += message_num_bytes;
+    consumer_num_bytes_ += message_num_bytes;
+  }
+
+  DCHECK_LE(consumer_num_bytes_, capacity_num_bytes());
+  // TODO(vtl): (End of copied code.)
+
+  set_producer_two_phase_max_num_bytes_written(0);
+  return MOJO_RESULT_OK;
+}
+
+HandleSignalsState RemoteConsumerDataPipeImpl::ProducerGetHandleSignalsState()
+    const {
+  HandleSignalsState rv;
+  if (consumer_open()) {
+    if (consumer_num_bytes_ < capacity_num_bytes() &&
+        !producer_in_two_phase_write())
+      rv.satisfied_signals |= MOJO_HANDLE_SIGNAL_WRITABLE;
+    rv.satisfiable_signals |= MOJO_HANDLE_SIGNAL_WRITABLE;
+  } else {
+    rv.satisfied_signals |= MOJO_HANDLE_SIGNAL_PEER_CLOSED;
+  }
+  rv.satisfiable_signals |= MOJO_HANDLE_SIGNAL_PEER_CLOSED;
+  return rv;
+}
+
+void RemoteConsumerDataPipeImpl::ProducerStartSerialize(
+    Channel* channel,
+    size_t* max_size,
+    size_t* max_platform_handles) {
+  *max_size = sizeof(SerializedDataPipeProducerDispatcher) +
+              channel->GetSerializedEndpointSize();
+  *max_platform_handles = 0;
+}
+
+bool RemoteConsumerDataPipeImpl::ProducerEndSerialize(
+    Channel* channel,
+    void* destination,
+    size_t* actual_size,
+    embedder::PlatformHandleVector* platform_handles) {
+  SerializedDataPipeProducerDispatcher* s =
+      static_cast<SerializedDataPipeProducerDispatcher*>(destination);
+  s->validated_options = validated_options();
+  void* destination_for_endpoint = static_cast<char*>(destination) +
+                                   sizeof(SerializedDataPipeProducerDispatcher);
+
+  if (!consumer_open()) {
+    // Case 1: The consumer is closed.
+    s->consumer_num_bytes = static_cast<size_t>(-1);
+    *actual_size = sizeof(SerializedDataPipeProducerDispatcher);
+    return true;
+  }
+
+  // Case 2: The consumer isn't closed. We pass |channel_endpoint| back to the
+  // |Channel|. There's no reason for us to continue to exist afterwards.
+
+  s->consumer_num_bytes = consumer_num_bytes_;
+  // Note: We don't use |port|.
+  scoped_refptr<ChannelEndpoint> channel_endpoint;
+  channel_endpoint.swap(channel_endpoint_);
+  channel->SerializeEndpointWithRemotePeer(destination_for_endpoint, nullptr,
+                                           channel_endpoint);
+  owner()->SetConsumerClosedNoLock();
+
+  *actual_size = sizeof(SerializedDataPipeProducerDispatcher) +
+                 channel->GetSerializedEndpointSize();
+  return true;
+}
+
+void RemoteConsumerDataPipeImpl::ConsumerClose() {
+  NOTREACHED();
+}
+
+MojoResult RemoteConsumerDataPipeImpl::ConsumerReadData(
+    UserPointer<void> /*elements*/,
+    UserPointer<uint32_t> /*num_bytes*/,
+    uint32_t /*max_num_bytes_to_read*/,
+    uint32_t /*min_num_bytes_to_read*/,
+    bool /*peek*/) {
+  NOTREACHED();
+  return MOJO_RESULT_INTERNAL;
+}
+
+MojoResult RemoteConsumerDataPipeImpl::ConsumerDiscardData(
+    UserPointer<uint32_t> /*num_bytes*/,
+    uint32_t /*max_num_bytes_to_discard*/,
+    uint32_t /*min_num_bytes_to_discard*/) {
+  NOTREACHED();
+  return MOJO_RESULT_INTERNAL;
+}
+
+MojoResult RemoteConsumerDataPipeImpl::ConsumerQueryData(
+    UserPointer<uint32_t> /*num_bytes*/) {
+  NOTREACHED();
+  return MOJO_RESULT_INTERNAL;
+}
+
+MojoResult RemoteConsumerDataPipeImpl::ConsumerBeginReadData(
+    UserPointer<const void*> /*buffer*/,
+    UserPointer<uint32_t> /*buffer_num_bytes*/,
+    uint32_t /*min_num_bytes_to_read*/) {
+  NOTREACHED();
+  return MOJO_RESULT_INTERNAL;
+}
+
+MojoResult RemoteConsumerDataPipeImpl::ConsumerEndReadData(
+    uint32_t /*num_bytes_read*/) {
+  NOTREACHED();
+  return MOJO_RESULT_INTERNAL;
+}
+
+HandleSignalsState RemoteConsumerDataPipeImpl::ConsumerGetHandleSignalsState()
+    const {
+  return HandleSignalsState();
+}
+
+void RemoteConsumerDataPipeImpl::ConsumerStartSerialize(
+    Channel* /*channel*/,
+    size_t* /*max_size*/,
+    size_t* /*max_platform_handles*/) {
+  NOTREACHED();
+}
+
+bool RemoteConsumerDataPipeImpl::ConsumerEndSerialize(
+    Channel* /*channel*/,
+    void* /*destination*/,
+    size_t* /*actual_size*/,
+    embedder::PlatformHandleVector* /*platform_handles*/) {
+  NOTREACHED();
+  return false;
+}
+
+bool RemoteConsumerDataPipeImpl::OnReadMessage(unsigned /*port*/,
+                                               MessageInTransit* message) {
+  // Always take ownership of the message. (This means that we should always
+  // return true.)
+  scoped_ptr<MessageInTransit> msg(message);
+
+  if (!ValidateIncomingMessage(element_num_bytes(), capacity_num_bytes(),
+                               consumer_num_bytes_, msg.get())) {
+    Disconnect();
+    return true;
+  }
+
+  const RemoteDataPipeAck* ack =
+      static_cast<const RemoteDataPipeAck*>(msg->bytes());
+  size_t num_bytes_consumed = ack->num_bytes_consumed;
+  consumer_num_bytes_ -= num_bytes_consumed;
+  return true;
+}
+
+void RemoteConsumerDataPipeImpl::OnDetachFromChannel(unsigned /*port*/) {
+  if (!consumer_open()) {
+    DCHECK(!channel_endpoint_);
+    return;
+  }
+
+  Disconnect();
+}
+
+void RemoteConsumerDataPipeImpl::EnsureBuffer() {
+  DCHECK(producer_open());
+  if (buffer_)
+    return;
+  buffer_.reset(static_cast<char*>(
+      base::AlignedAlloc(capacity_num_bytes(),
+                         GetConfiguration().data_pipe_buffer_alignment_bytes)));
+}
+
+void RemoteConsumerDataPipeImpl::DestroyBuffer() {
+#ifndef NDEBUG
+  // Scribble on the buffer to help detect use-after-frees. (This also helps the
+  // unit test detect certain bugs without needing ASAN or similar.)
+  if (buffer_)
+    memset(buffer_.get(), 0xcd, capacity_num_bytes());
+#endif
+  buffer_.reset();
+}
+
+void RemoteConsumerDataPipeImpl::Disconnect() {
+  DCHECK(consumer_open());
+  DCHECK(channel_endpoint_);
+  owner()->SetConsumerClosedNoLock();
+  channel_endpoint_->DetachFromClient();
+  channel_endpoint_ = nullptr;
+  DestroyBuffer();
+}
+
+}  // namespace system
+}  // namespace mojo