text: // Copyright 2013 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.
  
  // TODO(vtl): I currently potentially overflow in doing index calculations.
  // E.g., |start_index_| and |current_num_bytes_| fit into a |uint32_t|, but
  // their sum may not. This is bad and poses a security risk. (We're currently
  // saved by the limit on capacity -- the maximum size of the buffer, checked in
  // |DataPipe::ValidateOptions()|, is currently sufficiently small.)
  
  #include "third_party/mojo/src/mojo/edk/system/local_data_pipe_impl.h"
  
  #include <string.h>
  
  #include <algorithm>
  
  #include "base/logging.h"
  #include "base/memory/scoped_ptr.h"
  #include "third_party/mojo/src/mojo/edk/system/channel.h"
  #include "third_party/mojo/src/mojo/edk/system/configuration.h"
  #include "third_party/mojo/src/mojo/edk/system/data_pipe.h"
  #include "third_party/mojo/src/mojo/edk/system/message_in_transit.h"
  #include "third_party/mojo/src/mojo/edk/system/message_in_transit_queue.h"
  #include "third_party/mojo/src/mojo/edk/system/remote_consumer_data_pipe_impl.h"
  #include "third_party/mojo/src/mojo/edk/system/remote_producer_data_pipe_impl.h"
  
  namespace mojo {
  namespace system {
  
  // Assert some things about some things defined in data_pipe_impl.h (don't make
  // the assertions there, to avoid including message_in_transit.h).
  static_assert(MOJO_ALIGNOF(SerializedDataPipeConsumerDispatcher) ==
                    MessageInTransit::kMessageAlignment,
                "Wrong alignment");
  static_assert(sizeof(SerializedDataPipeConsumerDispatcher) %
                        MessageInTransit::kMessageAlignment ==
                    0,
                "Wrong size");
  
  LocalDataPipeImpl::LocalDataPipeImpl()
      : start_index_(0), current_num_bytes_(0) {
    // Note: |buffer_| is lazily allocated, since a common case will be that one
    // of the handles is immediately passed off to another process.
  }
  
  LocalDataPipeImpl::~LocalDataPipeImpl() {
  }
  
  void LocalDataPipeImpl::ProducerClose() {
    // 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();
    }
  }
  
  MojoResult LocalDataPipeImpl::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());
  
    if (min_num_bytes_to_write > capacity_num_bytes() - current_num_bytes_) {
      // Don't return "should wait" since you can't wait for a specified amount
      // of data.
      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() - current_num_bytes_);
    if (num_bytes_to_write == 0)
      return MOJO_RESULT_SHOULD_WAIT;
  
    // The amount we can write in our first copy.
    size_t num_bytes_to_write_first =
        std::min(num_bytes_to_write, GetMaxNumBytesToWrite());
    // Do the first (and possibly only) copy.
    size_t first_write_index =
        (start_index_ + current_num_bytes_) % capacity_num_bytes();
    EnsureBuffer();
    elements.GetArray(buffer_.get() + first_write_index,
                      num_bytes_to_write_first);
  
    if (num_bytes_to_write_first < num_bytes_to_write) {
      // The "second write index" is zero.
      elements.At(num_bytes_to_write_first)
          .GetArray(buffer_.get(), num_bytes_to_write - num_bytes_to_write_first);
    }
  
    current_num_bytes_ += num_bytes_to_write;
    DCHECK_LE(current_num_bytes_, capacity_num_bytes());
    num_bytes.Put(static_cast<uint32_t>(num_bytes_to_write));
    return MOJO_RESULT_OK;
  }
  
  MojoResult LocalDataPipeImpl::ProducerBeginWriteData(
      UserPointer<void*> buffer,
      UserPointer<uint32_t> buffer_num_bytes,
      uint32_t min_num_bytes_to_write) {
    DCHECK(consumer_open());
  
    // The index we need to start writing at.
    size_t write_index =
        (start_index_ + current_num_bytes_) % capacity_num_bytes();
  
    size_t max_num_bytes_to_write = GetMaxNumBytesToWrite();
    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() + write_index);
    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 LocalDataPipeImpl::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);
    current_num_bytes_ += num_bytes_written;
    DCHECK_LE(current_num_bytes_, capacity_num_bytes());
    set_producer_two_phase_max_num_bytes_written(0);
    return MOJO_RESULT_OK;
  }
  
  HandleSignalsState LocalDataPipeImpl::ProducerGetHandleSignalsState() const {
    HandleSignalsState rv;
    if (consumer_open()) {
      if (current_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 LocalDataPipeImpl::ProducerStartSerialize(Channel* channel,
                                                 size_t* max_size,
                                                 size_t* max_platform_handles) {
    *max_size = sizeof(SerializedDataPipeProducerDispatcher) +
                channel->GetSerializedEndpointSize();
    *max_platform_handles = 0;
  }
  
  bool LocalDataPipeImpl::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'll replace ourselves with a
    // |RemoteProducerDataPipeImpl|.
  
    s->consumer_num_bytes = current_num_bytes_;
    // Note: We don't use |port|.
    scoped_refptr<ChannelEndpoint> channel_endpoint =
        channel->SerializeEndpointWithLocalPeer(destination_for_endpoint, nullptr,
                                                owner(), 0);
    // Note: Keep |*this| alive until the end of this method, to make things
    // slightly easier on ourselves.
    scoped_ptr<DataPipeImpl> self(owner()->ReplaceImplNoLock(make_scoped_ptr(
        new RemoteProducerDataPipeImpl(channel_endpoint.get(), buffer_.Pass(),
                                       start_index_, current_num_bytes_))));
  
    *actual_size = sizeof(SerializedDataPipeProducerDispatcher) +
                   channel->GetSerializedEndpointSize();
    return true;
  }
  
  void LocalDataPipeImpl::ConsumerClose() {
    // If the producer is around and in a two-phase write, we have to keep the
    // buffer around. (We then don't free it until the producer is closed. This
    // could be rectified, but again seems like optimizing for the uncommon case.)
    if (!producer_open() || !producer_in_two_phase_write())
      DestroyBuffer();
    current_num_bytes_ = 0;
  }
  
  MojoResult LocalDataPipeImpl::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 copy.
    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 LocalDataPipeImpl::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 LocalDataPipeImpl::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 LocalDataPipeImpl::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 LocalDataPipeImpl::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 LocalDataPipeImpl::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 LocalDataPipeImpl::ConsumerStartSerialize(Channel* channel,
                                                 size_t* max_size,
                                                 size_t* max_platform_handles) {
    *max_size = sizeof(SerializedDataPipeConsumerDispatcher) +
                channel->GetSerializedEndpointSize();
    *max_platform_handles = 0;
  }
  
  bool LocalDataPipeImpl::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);
  
    size_t old_num_bytes = current_num_bytes_;
    MessageInTransitQueue message_queue;
    ConvertDataToMessages(buffer_.get(), &start_index_, &current_num_bytes_,
                          &message_queue);
    start_index_ = 0;
    current_num_bytes_ = 0;
  
    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'll replace ourselves with a
    // |RemoteConsumerDataPipeImpl|.
  
    // Note: We don't use |port|.
    scoped_refptr<ChannelEndpoint> channel_endpoint =
        channel->SerializeEndpointWithLocalPeer(destination_for_endpoint,
                                                &message_queue, owner(), 0);
    // Note: Keep |*this| alive until the end of this method, to make things
    // slightly easier on ourselves.
    scoped_ptr<DataPipeImpl> self(owner()->ReplaceImplNoLock(make_scoped_ptr(
        new RemoteConsumerDataPipeImpl(channel_endpoint.get(), old_num_bytes))));
  
    *actual_size = sizeof(SerializedDataPipeConsumerDispatcher) +
                   channel->GetSerializedEndpointSize();
    return true;
  }
  
  bool LocalDataPipeImpl::OnReadMessage(unsigned /*port*/,
                                        MessageInTransit* /*message*/) {
    NOTREACHED();
    return false;
  }
  
  void LocalDataPipeImpl::OnDetachFromChannel(unsigned /*port*/) {
    NOTREACHED();
  }
  
  void LocalDataPipeImpl::EnsureBuffer() {
    DCHECK(producer_open());
    if (buffer_)
      return;
    buffer_.reset(static_cast<char*>(
        base::AlignedAlloc(capacity_num_bytes(),
                           GetConfiguration().data_pipe_buffer_alignment_bytes)));
  }
  
  void LocalDataPipeImpl::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 LocalDataPipeImpl::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 LocalDataPipeImpl::GetMaxNumBytesToRead() {
    if (start_index_ + current_num_bytes_ > capacity_num_bytes())
      return capacity_num_bytes() - start_index_;
    return current_num_bytes_;
  }
  
  void LocalDataPipeImpl::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;
  }
  
  }  // namespace system
  }  // namespace mojo
  