Update mojo sdk to rev f68e697e389943cd9bf9652397312280e96b127a

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

Index: third_party/mojo/src/mojo/edk/system/remote_producer_data_pipe_impl.cc
diff --git a/third_party/mojo/src/mojo/edk/system/remote_producer_data_pipe_impl.cc b/third_party/mojo/src/mojo/edk/system/remote_producer_data_pipe_impl.cc
new file mode 100644
index 0000000000000000000000000000000000000000..7cd2447fd0079a5dfbd42df39089ba66fe6b80f8
--- /dev/null
+++ b/third_party/mojo/src/mojo/edk/system/remote_producer_data_pipe_impl.cc
@@ -0,0 +1,475 @@
+// 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_producer_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/message_in_transit_queue.h"
+#include "mojo/edk/system/remote_consumer_data_pipe_impl.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 current_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 messages.
+  if (message->subtype() != MessageInTransit::kSubtypeEndpointData) {
+    LOG(WARNING) << "Received message of unexpected subtype: "
+                 << message->subtype();
+    return false;
+  }
+
+  const size_t num_bytes = message->num_bytes();
+  const size_t max_num_bytes = capacity_num_bytes - current_num_bytes;
+  if (num_bytes > max_num_bytes) {
+    LOG(WARNING) << "Received too much data: " << num_bytes
+                 << " bytes (maximum: " << max_num_bytes << " bytes)";
+    return false;
+  }
+
+  if (num_bytes % element_num_bytes != 0) {
+    LOG(WARNING) << "Received data not a multiple of element size: "
+                 << num_bytes << " bytes (element size: " << element_num_bytes
+                 << " bytes)";
+    return false;
+  }
+
+  return true;
+}
+
+}  // namespace
+
+RemoteProducerDataPipeImpl::RemoteProducerDataPipeImpl(
+    ChannelEndpoint* channel_endpoint)
+    : channel_endpoint_(channel_endpoint),
+      start_index_(0),
+      current_num_bytes_(0) {
+  // Note: |buffer_| is lazily allocated.
+}
+
+RemoteProducerDataPipeImpl::RemoteProducerDataPipeImpl(
+    ChannelEndpoint* channel_endpoint,
+    scoped_ptr<char, base::AlignedFreeDeleter> buffer,
+    size_t start_index,
+    size_t current_num_bytes)
+    : channel_endpoint_(channel_endpoint),
+      buffer_(buffer.Pass()),
+      start_index_(start_index),
+      current_num_bytes_(current_num_bytes) {
+  DCHECK(buffer_ || !current_num_bytes);
+}
+
+// static
+bool RemoteProducerDataPipeImpl::ProcessMessagesFromIncomingEndpoint(
+    const MojoCreateDataPipeOptions& validated_options,
+    MessageInTransitQueue* messages,
+    scoped_ptr<char, base::AlignedFreeDeleter>* buffer,
+    size_t* buffer_num_bytes) {
+  DCHECK(!*buffer);  // Not wrong, but unlikely.
+
+  const size_t element_num_bytes = validated_options.element_num_bytes;
+  const size_t capacity_num_bytes = validated_options.capacity_num_bytes;
+
+  scoped_ptr<char, base::AlignedFreeDeleter> new_buffer(static_cast<char*>(
+      base::AlignedAlloc(capacity_num_bytes,
+                         GetConfiguration().data_pipe_buffer_alignment_bytes)));
+
+  size_t current_num_bytes = 0;
+  if (messages) {
+    while (!messages->IsEmpty()) {
+      scoped_ptr<MessageInTransit> message(messages->GetMessage());
+      if (!ValidateIncomingMessage(element_num_bytes, capacity_num_bytes,
+                                   current_num_bytes, message.get())) {
+        messages->Clear();
+        return false;
+      }
+
+      memcpy(new_buffer.get() + current_num_bytes, message->bytes(),
+             message->num_bytes());
+      current_num_bytes += message->num_bytes();
+    }
+  }
+
+  *buffer = new_buffer.Pass();
+  *buffer_num_bytes = current_num_bytes;
+  return true;
+}
+
+RemoteProducerDataPipeImpl::~RemoteProducerDataPipeImpl() {
+}
+
+void RemoteProducerDataPipeImpl::ProducerClose() {
+  NOTREACHED();
+}
+
+MojoResult RemoteProducerDataPipeImpl::ProducerWriteData(
+    UserPointer<const void> /*elements*/,
+    UserPointer<uint32_t> /*num_bytes*/,
+    uint32_t /*max_num_bytes_to_write*/,
+    uint32_t /*min_num_bytes_to_write*/) {
+  NOTREACHED();
+  return MOJO_RESULT_INTERNAL;
+}
+
+MojoResult RemoteProducerDataPipeImpl::ProducerBeginWriteData(
+    UserPointer<void*> /*buffer*/,
+    UserPointer<uint32_t> /*buffer_num_bytes*/,
+    uint32_t /*min_num_bytes_to_write*/) {
+  NOTREACHED();
+  return MOJO_RESULT_INTERNAL;
+}
+
+MojoResult RemoteProducerDataPipeImpl::ProducerEndWriteData(
+    uint32_t /*num_bytes_written*/) {
+  NOTREACHED();
+  return MOJO_RESULT_INTERNAL;
+}
+
+HandleSignalsState RemoteProducerDataPipeImpl::ProducerGetHandleSignalsState()
+    const {
+  return HandleSignalsState();
+}
+
+void RemoteProducerDataPipeImpl::ProducerStartSerialize(
+    Channel* /*channel*/,
+    size_t* /*max_size*/,
+    size_t* /*max_platform_handles*/) {
+  NOTREACHED();
+}
+
+bool RemoteProducerDataPipeImpl::ProducerEndSerialize(
+    Channel* /*channel*/,
+    void* /*destination*/,
+    size_t* /*actual_size*/,
+    embedder::PlatformHandleVector* /*platform_handles*/) {
+  NOTREACHED();
+  return false;
+}
+
+void RemoteProducerDataPipeImpl::ConsumerClose() {
+  if (producer_open())
+    Disconnect();
+  DestroyBuffer();
+  current_num_bytes_ = 0;
+}
+
+MojoResult RemoteProducerDataPipeImpl::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) {
+  DCHECK_EQ(max_num_bytes_to_read % element_num_bytes(), 0u);
+  DCHECK_EQ(min_num_bytes_to_read % element_num_bytes(), 0u);
+  DCHECK_GT(max_num_bytes_to_read, 0u);
+
+  if (min_num_bytes_to_read > current_num_bytes_) {
+    // Don't return "should wait" since you can't wait for a specified amount of
+    // data.
+    return producer_open() ? MOJO_RESULT_OUT_OF_RANGE
+                           : MOJO_RESULT_FAILED_PRECONDITION;
+  }
+
+  size_t num_bytes_to_read =
+      std::min(static_cast<size_t>(max_num_bytes_to_read), current_num_bytes_);
+  if (num_bytes_to_read == 0) {
+    return producer_open() ? MOJO_RESULT_SHOULD_WAIT
+                           : MOJO_RESULT_FAILED_PRECONDITION;
+  }
+
+  // The amount we can read in our first |memcpy()|.
+  size_t num_bytes_to_read_first =
+      std::min(num_bytes_to_read, GetMaxNumBytesToRead());
+  elements.PutArray(buffer_.get() + start_index_, num_bytes_to_read_first);
+
+  if (num_bytes_to_read_first < num_bytes_to_read) {
+    // The "second read index" is zero.
+    elements.At(num_bytes_to_read_first)
+        .PutArray(buffer_.get(), num_bytes_to_read - num_bytes_to_read_first);
+  }
+
+  if (!peek)
+    MarkDataAsConsumed(num_bytes_to_read);
+  num_bytes.Put(static_cast<uint32_t>(num_bytes_to_read));
+  return MOJO_RESULT_OK;
+}
+
+MojoResult RemoteProducerDataPipeImpl::ConsumerDiscardData(
+    UserPointer<uint32_t> num_bytes,
+    uint32_t max_num_bytes_to_discard,
+    uint32_t min_num_bytes_to_discard) {
+  DCHECK_EQ(max_num_bytes_to_discard % element_num_bytes(), 0u);
+  DCHECK_EQ(min_num_bytes_to_discard % element_num_bytes(), 0u);
+  DCHECK_GT(max_num_bytes_to_discard, 0u);
+
+  if (min_num_bytes_to_discard > current_num_bytes_) {
+    // Don't return "should wait" since you can't wait for a specified amount of
+    // data.
+    return producer_open() ? MOJO_RESULT_OUT_OF_RANGE
+                           : MOJO_RESULT_FAILED_PRECONDITION;
+  }
+
+  // Be consistent with other operations; error if no data available.
+  if (current_num_bytes_ == 0) {
+    return producer_open() ? MOJO_RESULT_SHOULD_WAIT
+                           : MOJO_RESULT_FAILED_PRECONDITION;
+  }
+
+  size_t num_bytes_to_discard = std::min(
+      static_cast<size_t>(max_num_bytes_to_discard), current_num_bytes_);
+  MarkDataAsConsumed(num_bytes_to_discard);
+  num_bytes.Put(static_cast<uint32_t>(num_bytes_to_discard));
+  return MOJO_RESULT_OK;
+}
+
+MojoResult RemoteProducerDataPipeImpl::ConsumerQueryData(
+    UserPointer<uint32_t> num_bytes) {
+  // Note: This cast is safe, since the capacity fits into a |uint32_t|.
+  num_bytes.Put(static_cast<uint32_t>(current_num_bytes_));
+  return MOJO_RESULT_OK;
+}
+
+MojoResult RemoteProducerDataPipeImpl::ConsumerBeginReadData(
+    UserPointer<const void*> buffer,
+    UserPointer<uint32_t> buffer_num_bytes,
+    uint32_t min_num_bytes_to_read) {
+  size_t max_num_bytes_to_read = GetMaxNumBytesToRead();
+  if (min_num_bytes_to_read > max_num_bytes_to_read) {
+    // Don't return "should wait" since you can't wait for a specified amount of
+    // data.
+    return producer_open() ? MOJO_RESULT_OUT_OF_RANGE
+                           : MOJO_RESULT_FAILED_PRECONDITION;
+  }
+
+  // Don't go into a two-phase read if there's no data.
+  if (max_num_bytes_to_read == 0) {
+    return producer_open() ? MOJO_RESULT_SHOULD_WAIT
+                           : MOJO_RESULT_FAILED_PRECONDITION;
+  }
+
+  buffer.Put(buffer_.get() + start_index_);
+  buffer_num_bytes.Put(static_cast<uint32_t>(max_num_bytes_to_read));
+  set_consumer_two_phase_max_num_bytes_read(
+      static_cast<uint32_t>(max_num_bytes_to_read));
+  return MOJO_RESULT_OK;
+}
+
+MojoResult RemoteProducerDataPipeImpl::ConsumerEndReadData(
+    uint32_t num_bytes_read) {
+  DCHECK_LE(num_bytes_read, consumer_two_phase_max_num_bytes_read());
+  DCHECK_EQ(num_bytes_read % element_num_bytes(), 0u);
+  DCHECK_LE(start_index_ + num_bytes_read, capacity_num_bytes());
+  MarkDataAsConsumed(num_bytes_read);
+  set_consumer_two_phase_max_num_bytes_read(0);
+  return MOJO_RESULT_OK;
+}
+
+HandleSignalsState RemoteProducerDataPipeImpl::ConsumerGetHandleSignalsState()
+    const {
+  HandleSignalsState rv;
+  if (current_num_bytes_ > 0) {
+    if (!consumer_in_two_phase_read())
+      rv.satisfied_signals |= MOJO_HANDLE_SIGNAL_READABLE;
+    rv.satisfiable_signals |= MOJO_HANDLE_SIGNAL_READABLE;
+  } else if (producer_open()) {
+    rv.satisfiable_signals |= MOJO_HANDLE_SIGNAL_READABLE;
+  }
+  if (!producer_open())
+    rv.satisfied_signals |= MOJO_HANDLE_SIGNAL_PEER_CLOSED;
+  rv.satisfiable_signals |= MOJO_HANDLE_SIGNAL_PEER_CLOSED;
+  return rv;
+}
+
+void RemoteProducerDataPipeImpl::ConsumerStartSerialize(
+    Channel* channel,
+    size_t* max_size,
+    size_t* max_platform_handles) {
+  *max_size = sizeof(SerializedDataPipeConsumerDispatcher) +
+              channel->GetSerializedEndpointSize();
+  *max_platform_handles = 0;
+}
+
+bool RemoteProducerDataPipeImpl::ConsumerEndSerialize(
+    Channel* channel,
+    void* destination,
+    size_t* actual_size,
+    embedder::PlatformHandleVector* platform_handles) {
+  SerializedDataPipeConsumerDispatcher* s =
+      static_cast<SerializedDataPipeConsumerDispatcher*>(destination);
+  s->validated_options = validated_options();
+  void* destination_for_endpoint = static_cast<char*>(destination) +
+                                   sizeof(SerializedDataPipeConsumerDispatcher);
+
+  MessageInTransitQueue message_queue;
+  ConvertDataToMessages(buffer_.get(), &start_index_, &current_num_bytes_,
+                        &message_queue);
+
+  if (!producer_open()) {
+    // Case 1: The producer is closed.
+    channel->SerializeEndpointWithClosedPeer(destination_for_endpoint,
+                                             &message_queue);
+    *actual_size = sizeof(SerializedDataPipeConsumerDispatcher) +
+                   channel->GetSerializedEndpointSize();
+    return true;
+  }
+
+  // Case 2: The producer isn't closed. We pass |channel_endpoint| back to the
+  // |Channel|. There's no reason for us to continue to exist afterwards.
+
+  // Note: We don't use |port|.
+  scoped_refptr<ChannelEndpoint> channel_endpoint;
+  channel_endpoint.swap(channel_endpoint_);
+  channel->SerializeEndpointWithRemotePeer(destination_for_endpoint,
+                                           &message_queue, channel_endpoint);
+  owner()->SetProducerClosedNoLock();
+
+  *actual_size = sizeof(SerializedDataPipeConsumerDispatcher) +
+                 channel->GetSerializedEndpointSize();
+  return true;
+}
+
+bool RemoteProducerDataPipeImpl::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 (!producer_open()) {
+    DCHECK(!channel_endpoint_);
+    return true;
+  }
+
+  if (!ValidateIncomingMessage(element_num_bytes(), capacity_num_bytes(),
+                               current_num_bytes_, msg.get())) {
+    Disconnect();
+    return true;
+  }
+
+  size_t num_bytes = msg->num_bytes();
+  // The amount we can write in our first copy.
+  size_t num_bytes_to_copy_first = std::min(num_bytes, GetMaxNumBytesToWrite());
+  // Do the first (and possibly only) copy.
+  size_t first_write_index =
+      (start_index_ + current_num_bytes_) % capacity_num_bytes();
+  EnsureBuffer();
+  memcpy(buffer_.get() + first_write_index, msg->bytes(),
+         num_bytes_to_copy_first);
+
+  if (num_bytes_to_copy_first < num_bytes) {
+    // The "second write index" is zero.
+    memcpy(buffer_.get(),
+           static_cast<const char*>(msg->bytes()) + num_bytes_to_copy_first,
+           num_bytes - num_bytes_to_copy_first);
+  }
+
+  current_num_bytes_ += num_bytes;
+  DCHECK_LE(current_num_bytes_, capacity_num_bytes());
+  return true;
+}
+
+void RemoteProducerDataPipeImpl::OnDetachFromChannel(unsigned /*port*/) {
+  if (!producer_open()) {
+    DCHECK(!channel_endpoint_);
+    return;
+  }
+
+  Disconnect();
+}
+
+void RemoteProducerDataPipeImpl::EnsureBuffer() {
+  DCHECK(producer_open());
+  if (buffer_)
+    return;
+  buffer_.reset(static_cast<char*>(
+      base::AlignedAlloc(capacity_num_bytes(),
+                         GetConfiguration().data_pipe_buffer_alignment_bytes)));
+}
+
+void RemoteProducerDataPipeImpl::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();
+}
+
+size_t RemoteProducerDataPipeImpl::GetMaxNumBytesToWrite() {
+  size_t next_index = start_index_ + current_num_bytes_;
+  if (next_index >= capacity_num_bytes()) {
+    next_index %= capacity_num_bytes();
+    DCHECK_GE(start_index_, next_index);
+    DCHECK_EQ(start_index_ - next_index,
+              capacity_num_bytes() - current_num_bytes_);
+    return start_index_ - next_index;
+  }
+  return capacity_num_bytes() - next_index;
+}
+
+size_t RemoteProducerDataPipeImpl::GetMaxNumBytesToRead() {
+  if (start_index_ + current_num_bytes_ > capacity_num_bytes())
+    return capacity_num_bytes() - start_index_;
+  return current_num_bytes_;
+}
+
+void RemoteProducerDataPipeImpl::MarkDataAsConsumed(size_t num_bytes) {
+  DCHECK_LE(num_bytes, current_num_bytes_);
+  start_index_ += num_bytes;
+  start_index_ %= capacity_num_bytes();
+  current_num_bytes_ -= num_bytes;
+
+  if (!producer_open()) {
+    DCHECK(!channel_endpoint_);
+    return;
+  }
+
+  RemoteDataPipeAck ack_data = {};
+  ack_data.num_bytes_consumed = static_cast<uint32_t>(num_bytes);
+  scoped_ptr<MessageInTransit> message(
+      new MessageInTransit(MessageInTransit::kTypeEndpoint,
+                           MessageInTransit::kSubtypeEndpointDataPipeAck,
+                           static_cast<uint32_t>(sizeof(ack_data)), &ack_data));
+  if (!channel_endpoint_->EnqueueMessage(message.Pass()))
+    Disconnect();
+}
+
+void RemoteProducerDataPipeImpl::Disconnect() {
+  DCHECK(producer_open());
+  DCHECK(channel_endpoint_);
+  owner()->SetProducerClosedNoLock();
+  channel_endpoint_->DetachFromClient();
+  channel_endpoint_ = nullptr;
+  // If the consumer is still open and we still have data, we have to keep the
+  // buffer around. Currently, we won't free it even if it empties later. (We
+  // could do this -- requiring a check on every read -- but that seems to be
+  // optimizing for the uncommon case.)
+  if (!consumer_open() || !current_num_bytes_) {
+    // Note: There can only be a two-phase *read* (by the consumer) if we still
+    // have data.
+    DCHECK(!consumer_in_two_phase_read());
+    DestroyBuffer();
+  }
+}
+
+}  // namespace system
+}  // namespace mojo