mojo/edk/system/raw_channel.cc

// Copyright 2014 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/raw_channel.h"

#include <stddef.h>
#include <stdint.h>
#include <string.h>
#include <algorithm>
#include <utility>

#include "base/bind.h"
#include "base/location.h"
#include "base/logging.h"
#include "base/message_loop/message_loop.h"
#include "mojo/edk/embedder/embedder_internal.h"
#include "mojo/edk/system/configuration.h"
#include "mojo/edk/system/message_in_transit.h"
#include "mojo/edk/system/transport_data.h"

namespace mojo {
namespace edk {

const size_t kReadSize = 4096;

// RawChannel::ReadBuffer ------------------------------------------------------

RawChannel::ReadBuffer::ReadBuffer() : buffer_(kReadSize), num_valid_bytes_(0) {
}

RawChannel::ReadBuffer::~ReadBuffer() {
}

void RawChannel::ReadBuffer::GetBuffer(char** addr, size_t* size) {
  DCHECK_GE(buffer_.size(), num_valid_bytes_ + kReadSize);
  *addr = &buffer_[0] + num_valid_bytes_;
  *size = kReadSize;
}

// RawChannel::WriteBuffer -----------------------------------------------------

RawChannel::WriteBuffer::WriteBuffer()
    : serialized_platform_handle_size_(0),
      platform_handles_offset_(0),
      data_offset_(0) {
}

RawChannel::WriteBuffer::~WriteBuffer() {
  message_queue_.Clear();
}

bool RawChannel::WriteBuffer::HavePlatformHandlesToSend() const {
  if (message_queue_.IsEmpty())
    return false;

  const TransportData* transport_data =
      message_queue_.PeekMessage()->transport_data();
  if (!transport_data)
    return false;

  const PlatformHandleVector* all_platform_handles =
      transport_data->platform_handles();
  if (!all_platform_handles) {
    DCHECK_EQ(platform_handles_offset_, 0u);
    return false;
  }
  if (platform_handles_offset_ >= all_platform_handles->size()) {
    DCHECK_EQ(platform_handles_offset_, all_platform_handles->size());
    return false;
  }

  return true;
}

void RawChannel::WriteBuffer::GetPlatformHandlesToSend(
    size_t* num_platform_handles,
    PlatformHandle** platform_handles,
    void** serialization_data) {
  DCHECK(HavePlatformHandlesToSend());

  MessageInTransit* message = message_queue_.PeekMessage();
  TransportData* transport_data = message->transport_data();
  PlatformHandleVector* all_platform_handles =
      transport_data->platform_handles();
  *num_platform_handles =
      all_platform_handles->size() - platform_handles_offset_;
  *platform_handles = &(*all_platform_handles)[platform_handles_offset_];

  if (serialized_platform_handle_size_ > 0) {
    size_t serialization_data_offset =
        transport_data->platform_handle_table_offset();
    serialization_data_offset +=
        platform_handles_offset_ * serialized_platform_handle_size_;
    *serialization_data = static_cast<char*>(transport_data->buffer()) +
                          serialization_data_offset;
  } else {
    *serialization_data = nullptr;
  }
}

void RawChannel::WriteBuffer::GetBuffers(std::vector<Buffer>* buffers) {
  buffers->clear();

  if (message_queue_.IsEmpty())
    return;

  const MessageInTransit* message = message_queue_.PeekMessage();
  if (message->type() == MessageInTransit::Type::RAW_MESSAGE) {
    // These are already-serialized messages so we don't want to write another
    // header as they include that.
    if (data_offset_ == 0) {
      size_t header_size = message->total_size() - message->num_bytes();
      data_offset_ = header_size;
    }
  }

  DCHECK_LT(data_offset_, message->total_size());
  size_t bytes_to_write = message->total_size() - data_offset_;

  size_t transport_data_buffer_size =
      message->transport_data() ? message->transport_data()->buffer_size() : 0;

  if (!transport_data_buffer_size) {
    // Only write from the main buffer.
    DCHECK_LT(data_offset_, message->main_buffer_size());
    DCHECK_LE(bytes_to_write, message->main_buffer_size());
    Buffer buffer = {
        static_cast<const char*>(message->main_buffer()) + data_offset_,
        bytes_to_write};

    buffers->push_back(buffer);
    return;
  }

  if (data_offset_ >= message->main_buffer_size()) {
    // Only write from the transport data buffer.
    DCHECK_LT(data_offset_ - message->main_buffer_size(),
              transport_data_buffer_size);
    DCHECK_LE(bytes_to_write, transport_data_buffer_size);
    Buffer buffer = {
        static_cast<const char*>(message->transport_data()->buffer()) +
            (data_offset_ - message->main_buffer_size()),
        bytes_to_write};

    buffers->push_back(buffer);
    return;
  }

  // TODO(vtl): We could actually send out buffers from multiple messages, with
  // the "stopping" condition being reaching a message with platform handles
  // attached.

  // Write from both buffers.
  DCHECK_EQ(bytes_to_write, message->main_buffer_size() - data_offset_ +
                                transport_data_buffer_size);
  Buffer buffer1 = {
      static_cast<const char*>(message->main_buffer()) + data_offset_,
      message->main_buffer_size() - data_offset_};
  buffers->push_back(buffer1);
  Buffer buffer2 = {
      static_cast<const char*>(message->transport_data()->buffer()),
      transport_data_buffer_size};
  buffers->push_back(buffer2);
}

// RawChannel ------------------------------------------------------------------

RawChannel::RawChannel()
    : delegate_(nullptr),
      error_occurred_(false),
      calling_delegate_(false),
      write_ready_(false),
      write_stopped_(false),
      pending_write_error_(false),
      initialized_(false),
      weak_ptr_factory_(this) {
  read_buffer_.reset(new ReadBuffer);
  write_buffer_.reset(new WriteBuffer());
}

RawChannel::~RawChannel() {
  DCHECK(!read_buffer_);
  DCHECK(!write_buffer_);
}

void RawChannel::Init(Delegate* delegate) {
  DCHECK(delegate);

  base::AutoLock read_locker(read_lock_);
  // Solves race where initialiing on io thread while main thread is serializing
  // this channel and releases handle.
  base::AutoLock locker(write_lock_);

  DCHECK(!delegate_);
  delegate_ = delegate;

  if (read_buffer_->num_valid_bytes_ ||
      !write_buffer_->message_queue_.IsEmpty()) {
    LazyInitialize();
  }
}

void RawChannel::EnsureLazyInitialized() {
  if (!initialized_) {
    internal::g_io_thread_task_runner->PostTask(
        FROM_HERE,
        base::Bind(&RawChannel::LockAndCallLazyInitialize,
        weak_ptr_factory_.GetWeakPtr()));
  }
}

void RawChannel::LockAndCallLazyInitialize() {
  base::AutoLock read_locker(read_lock_);
  base::AutoLock locker(write_lock_);
  LazyInitialize();
}

void RawChannel::LazyInitialize() {
  read_lock_.AssertAcquired();
  write_lock_.AssertAcquired();
  DCHECK(internal::g_io_thread_task_runner->RunsTasksOnCurrentThread());
  if (initialized_)
    return;
  initialized_ = true;
  base::MessageLoop::current()->AddDestructionObserver(this);

  OnInit();

  if (read_buffer_->num_valid_bytes_) {
    // We had serialized read buffer data through SetSerializedData call.
    // Make sure we read messages out of it now, otherwise the delegate won't
    // get notified if no other data gets written to the pipe.
    // Although this means that we can call back synchronously into the caller,
    // that's easier than posting a task to do this. That is because if we post
    // a task, a pending read could have started and we wouldn't be able to move
    // the read buffer since it can be in use by the OS in an async operation.
    bool did_dispatch_message = false;
    bool stop_dispatching = false;
    DispatchMessages(&did_dispatch_message, &stop_dispatching);
  }

  IOResult io_result = ScheduleRead();
  if (io_result != IO_PENDING) {
    // This will notify the delegate about the read failure. Although we're on
    // the I/O thread, don't call it in the nested context.
    internal::g_io_thread_task_runner->PostTask(
        FROM_HERE, base::Bind(&RawChannel::CallOnReadCompleted,
                              weak_ptr_factory_.GetWeakPtr(), io_result, 0));
  }
  // Note: |ScheduleRead()| failure is treated as a read failure (by notifying
  // the delegate), not an initialization failure.

  write_ready_ = true;
  write_buffer_->serialized_platform_handle_size_ =
      GetSerializedPlatformHandleSize();
  if (!write_buffer_->message_queue_.IsEmpty())
    SendQueuedMessagesNoLock();
}

void RawChannel::Shutdown() {
  DCHECK(internal::g_io_thread_task_runner->RunsTasksOnCurrentThread());

  weak_ptr_factory_.InvalidateWeakPtrs();
  // Reset the delegate so that it won't receive further calls.
  delegate_ = nullptr;
  if (calling_delegate_) {
    internal::g_io_thread_task_runner->PostTask(
        FROM_HERE,
        base::Bind(&RawChannel::Shutdown, weak_ptr_factory_.GetWeakPtr()));
    return;
  }

  bool empty = false;
  {
    base::AutoLock locker(write_lock_);
    empty = write_buffer_->message_queue_.IsEmpty();
  }

  // Normally, we want to flush any pending writes before shutting down. This
  // doesn't apply when 1) we don't have a handle (for obvious reasons),
  // 2) we have a read or write error before (doesn't matter which), or 3) when
  // there are no pending messages to be written.
  if (!IsHandleValid() || error_occurred_ || empty) {
    {
      base::AutoLock read_locker(read_lock_);
      base::AutoLock locker(write_lock_);
      OnShutdownNoLock(std::move(read_buffer_), std::move(write_buffer_));
    }

    if (initialized_) {
      base::MessageLoop::current()->RemoveDestructionObserver(this);
    }
    delete this;
    return;
  }

  base::AutoLock read_locker(read_lock_);
  base::AutoLock locker(write_lock_);
  DCHECK(read_buffer_->IsEmpty()) <<
      "RawChannel::Shutdown called but there is pending data to be read";

  write_stopped_ = true;
}

ScopedPlatformHandle RawChannel::ReleaseHandle(
    std::vector<char>* serialized_read_buffer,
    std::vector<char>* serialized_write_buffer,
    std::vector<int>* serialized_read_fds,
    std::vector<int>* serialized_write_fds,
    bool* write_error) {
  ScopedPlatformHandle rv;
  *write_error = false;
  {
    base::AutoLock read_locker(read_lock_);
    base::AutoLock locker(write_lock_);
    rv = ReleaseHandleNoLock(serialized_read_buffer,
                             serialized_write_buffer,
                             serialized_read_fds,
                             serialized_write_fds,
                             write_error);
    delegate_ = nullptr;
    internal::g_io_thread_task_runner->PostTask(
        FROM_HERE,
        base::Bind(&RawChannel::Shutdown, weak_ptr_factory_.GetWeakPtr()));
  }

  return rv;
}

// Reminder: This must be thread-safe.
bool RawChannel::WriteMessage(scoped_ptr<MessageInTransit> message) {
  DCHECK(message);
  EnsureLazyInitialized();
  base::AutoLock locker(write_lock_);
  if (write_stopped_)
    return false;

  bool queue_was_empty = write_buffer_->message_queue_.IsEmpty();
  EnqueueMessageNoLock(std::move(message));
  if (queue_was_empty && write_ready_)
    return SendQueuedMessagesNoLock();

  return true;
}

bool RawChannel::SendQueuedMessagesNoLock() {
  DCHECK_EQ(write_buffer_->data_offset_, 0u);

  size_t platform_handles_written = 0;
  size_t bytes_written = 0;
  IOResult io_result = WriteNoLock(&platform_handles_written, &bytes_written);
  if (io_result == IO_PENDING)
    return true;

  bool result = OnWriteCompletedInternalNoLock(
      io_result, platform_handles_written, bytes_written);
  if (!result) {
    // Even if we're on the I/O thread, don't call |OnError()| in the nested
    // context.
    pending_write_error_ = true;
    internal::g_io_thread_task_runner->PostTask(
        FROM_HERE,
        base::Bind(&RawChannel::LockAndCallOnError,
                   weak_ptr_factory_.GetWeakPtr(),
                   Delegate::ERROR_WRITE));
  }

  return result;
}

void RawChannel::SetSerializedData(
    char* serialized_read_buffer, size_t serialized_read_buffer_size,
    char* serialized_write_buffer, size_t serialized_write_buffer_size,
    std::vector<int>* serialized_read_fds,
    std::vector<int>* serialized_write_fds) {
  base::AutoLock locker(read_lock_);

#if defined(OS_POSIX)
  SetSerializedFDs(serialized_read_fds, serialized_write_fds);
#endif

  if (serialized_read_buffer_size) {
    // TODO(jam): copy power of 2 algorithm below? or share.
    read_buffer_->buffer_.resize(serialized_read_buffer_size + kReadSize);
    memcpy(&read_buffer_->buffer_[0], serialized_read_buffer,
           serialized_read_buffer_size);
    read_buffer_->num_valid_bytes_ = serialized_read_buffer_size;
  }

  if (serialized_write_buffer_size) {
    size_t max_message_num_bytes = GetConfiguration().max_message_num_bytes;

    uint32_t offset = 0;
    while (offset < serialized_write_buffer_size) {
      uint32_t message_num_bytes =
          std::min(static_cast<uint32_t>(max_message_num_bytes),
                   static_cast<uint32_t>(serialized_write_buffer_size) -
                       offset);
      scoped_ptr<MessageInTransit> message(new MessageInTransit(
          MessageInTransit::Type::RAW_MESSAGE, message_num_bytes,
          static_cast<const char*>(serialized_write_buffer) + offset));
      write_buffer_->message_queue_.AddMessage(std::move(message));
      offset += message_num_bytes;
    }
  }
}

void RawChannel::OnReadCompletedNoLock(IOResult io_result, size_t bytes_read) {
  DCHECK(internal::g_io_thread_task_runner->RunsTasksOnCurrentThread());
  read_lock_.AssertAcquired();
  // Keep reading data in a loop, and dispatch messages if enough data is
  // received. Exit the loop if any of the following happens:
  //   - one or more messages were dispatched;
  //   - the last read failed, was a partial read or would block;
  //   - |Shutdown()| was called.
  do {
    switch (io_result) {
      case IO_SUCCEEDED:
        break;
      case IO_FAILED_SHUTDOWN:
      case IO_FAILED_BROKEN:
      case IO_FAILED_UNKNOWN:
        CallOnError(ReadIOResultToError(io_result));
        return;  // |this| may have been destroyed in |CallOnError()|.
      case IO_PENDING:
        NOTREACHED();
        return;
    }

    read_buffer_->num_valid_bytes_ += bytes_read;

    // Dispatch all the messages that we can.
    bool did_dispatch_message = false;
    bool stop_dispatching = false;
    DispatchMessages(&did_dispatch_message, &stop_dispatching);
    if (stop_dispatching)
      return;

    if (read_buffer_->buffer_.size() - read_buffer_->num_valid_bytes_ <
        kReadSize) {
      // Use power-of-2 buffer sizes.
      // TODO(vtl): Make sure the buffer doesn't get too large (and enforce the
      // maximum message size to whatever extent necessary).
      // TODO(vtl): We may often be able to peek at the header and get the real
      // required extra space (which may be much bigger than |kReadSize|).
      size_t new_size = std::max(read_buffer_->buffer_.size(), kReadSize);
      while (new_size < read_buffer_->num_valid_bytes_ + kReadSize)
        new_size *= 2;

      // TODO(vtl): It's suboptimal to zero out the fresh memory.
      read_buffer_->buffer_.resize(new_size, 0);
    }

    // (1) If we dispatched any messages, stop reading for now (and let the
    // message loop do its thing for another round).
    // TODO(vtl): Is this the behavior we want? (Alternatives: i. Dispatch only
    // a single message. Risks: slower, more complex if we want to avoid lots of
    // copying. ii. Keep reading until there's no more data and dispatch all the
    // messages we can. Risks: starvation of other users of the message loop.)
    // (2) If we didn't max out |kReadSize|, stop reading for now.
    bool schedule_for_later = did_dispatch_message || bytes_read < kReadSize;
    bytes_read = 0;
    io_result = schedule_for_later ? ScheduleRead() : Read(&bytes_read);
  } while (io_result != IO_PENDING);
}

void RawChannel::OnWriteCompletedNoLock(IOResult io_result,
                                        size_t platform_handles_written,
                                        size_t bytes_written) {
  DCHECK(internal::g_io_thread_task_runner->RunsTasksOnCurrentThread());
  write_lock_.AssertAcquired();
  DCHECK_NE(io_result, IO_PENDING);

  bool did_fail = !OnWriteCompletedInternalNoLock(
      io_result, platform_handles_written, bytes_written);
  if (did_fail) {
    // Don't want to call the delegate with the current callstack for two
    // reasons:
    // 1) We already have write_lock_ acquired, and calling the delegate means
    // also acquiring the read lock. We need to acquire read and then write to
    // avoid deadlocks.
    // 2) We shouldn't call the delegate with write_lock acquired, since the
    // delegate could be calling WriteMessage and that can cause deadlocks.
    pending_write_error_ = true;
    internal::g_io_thread_task_runner->PostTask(
        FROM_HERE,
        base::Bind(&RawChannel::LockAndCallOnError,
                   weak_ptr_factory_.GetWeakPtr(),
                   Delegate::ERROR_WRITE));
  }
}

void RawChannel::SerializeReadBuffer(size_t additional_bytes_read,
                                     std::vector<char>* buffer) {
  read_lock_.AssertAcquired();
  read_buffer_->num_valid_bytes_ += additional_bytes_read;
  read_buffer_->buffer_.resize(read_buffer_->num_valid_bytes_);
  read_buffer_->buffer_.swap(*buffer);
  read_buffer_->num_valid_bytes_ = 0;
}

void RawChannel::SerializeWriteBuffer(
    size_t additional_bytes_written,
    size_t additional_platform_handles_written,
    std::vector<char>* buffer,
    std::vector<int>* fds) {
  write_lock_.AssertAcquired();
  if (write_buffer_->IsEmpty()) {
    DCHECK_EQ(0u, additional_bytes_written);
    DCHECK_EQ(0u, additional_platform_handles_written);
    return;
  }

  UpdateWriteBuffer(
      additional_platform_handles_written, additional_bytes_written);
  while (!write_buffer_->message_queue_.IsEmpty()) {
    SerializePlatformHandles(fds);
    std::vector<WriteBuffer::Buffer> buffers;
    write_buffer_no_lock()->GetBuffers(&buffers);
    for (size_t i = 0; i < buffers.size(); ++i) {
      buffer->insert(buffer->end(), buffers[i].addr,
                     buffers[i].addr + buffers[i].size);
    }
    write_buffer_->message_queue_.DiscardMessage();
  }
}

void RawChannel::EnqueueMessageNoLock(scoped_ptr<MessageInTransit> message) {
  write_lock_.AssertAcquired();
  write_buffer_->message_queue_.AddMessage(std::move(message));
}

bool RawChannel::OnReadMessageForRawChannel(
    const MessageInTransit::View& message_view) {
  LOG(ERROR) << "Invalid control message (type " << message_view.type()
             << ")";
  return false;
}

RawChannel::Delegate::Error RawChannel::ReadIOResultToError(
    IOResult io_result) {
  switch (io_result) {
    case IO_FAILED_SHUTDOWN:
      return Delegate::ERROR_READ_SHUTDOWN;
    case IO_FAILED_BROKEN:
      return Delegate::ERROR_READ_BROKEN;
    case IO_FAILED_UNKNOWN:
      return Delegate::ERROR_READ_UNKNOWN;
    case IO_SUCCEEDED:
    case IO_PENDING:
      NOTREACHED();
      break;
  }
  return Delegate::ERROR_READ_UNKNOWN;
}

void RawChannel::CallOnError(Delegate::Error error) {
  DCHECK(internal::g_io_thread_task_runner->RunsTasksOnCurrentThread());
  read_lock_.AssertAcquired();
  error_occurred_ = true;
  if (delegate_) {
    DCHECK(!calling_delegate_);
    calling_delegate_ = true;
    delegate_->OnError(error);
    calling_delegate_ = false;
  } else {
    // We depend on delegate to delete since it could be waiting to call
    // ReleaseHandle.
    internal::g_io_thread_task_runner->PostTask(
        FROM_HERE,
        base::Bind(&RawChannel::Shutdown, weak_ptr_factory_.GetWeakPtr()));
  }
}

void RawChannel::LockAndCallOnError(Delegate::Error error) {
  base::AutoLock locker(read_lock_);
  CallOnError(error);
}

bool RawChannel::OnWriteCompletedInternalNoLock(IOResult io_result,
                                                size_t platform_handles_written,
                                                size_t bytes_written) {
  write_lock_.AssertAcquired();

  DCHECK(!write_buffer_->message_queue_.IsEmpty());

  if (io_result == IO_SUCCEEDED) {
    UpdateWriteBuffer(platform_handles_written, bytes_written);
    if (write_buffer_->message_queue_.IsEmpty()) {
      if (!delegate_) {
        // Shutdown must have been called and we were waiting to flush all
        // pending writes. Now we're done.
        internal::g_io_thread_task_runner->PostTask(
            FROM_HERE,
            base::Bind(&RawChannel::Shutdown, weak_ptr_factory_.GetWeakPtr()));
      }
      return true;
    }

    // Schedule the next write.
    io_result = ScheduleWriteNoLock();
    if (io_result == IO_PENDING)
      return true;
    DCHECK_NE(io_result, IO_SUCCEEDED);
  }

  write_stopped_ = true;
  write_buffer_->message_queue_.Clear();
  write_buffer_->platform_handles_offset_ = 0;
  write_buffer_->data_offset_ = 0;
  return false;
}

void RawChannel::DispatchMessages(bool* did_dispatch_message,
                                  bool* stop_dispatching) {
  *did_dispatch_message = false;
  *stop_dispatching = false;
  // Tracks the offset of the first undispatched message in |read_buffer_|.
  // Currently, we copy data to ensure that this is zero at the beginning.
  size_t read_buffer_start = 0;
  size_t remaining_bytes = read_buffer_->num_valid_bytes_;
  size_t message_size;
  // Note that we rely on short-circuit evaluation here:
  //   - |read_buffer_start| may be an invalid index into
  //     |read_buffer_->buffer_| if |remaining_bytes| is zero.
  //   - |message_size| is only valid if |GetNextMessageSize()| returns true.
  // TODO(vtl): Use |message_size| more intelligently (e.g., to request the
  // next read).
  // TODO(vtl): Validate that |message_size| is sane.
  while (remaining_bytes > 0 && MessageInTransit::GetNextMessageSize(
                                    &read_buffer_->buffer_[read_buffer_start],
                                    remaining_bytes, &message_size) &&
          remaining_bytes >= message_size) {
    MessageInTransit::View message_view(
        message_size, &read_buffer_->buffer_[read_buffer_start]);
    DCHECK_EQ(message_view.total_size(), message_size);

    const char* error_message = nullptr;
    if (!message_view.IsValid(GetSerializedPlatformHandleSize(),
                              &error_message)) {
      DCHECK(error_message);
      LOG(ERROR) << "Received invalid message: " << error_message;
      CallOnError(Delegate::ERROR_READ_BAD_MESSAGE);
      *stop_dispatching = true;
      return;  // |this| may have been destroyed in |CallOnError()|.
    }

    if (message_view.type() != MessageInTransit::Type::MESSAGE) {
      if (!OnReadMessageForRawChannel(message_view)) {
        CallOnError(Delegate::ERROR_READ_BAD_MESSAGE);
        *stop_dispatching = true;
        return;  // |this| may have been destroyed in |CallOnError()|.
      }
    } else {
      ScopedPlatformHandleVectorPtr platform_handles;
      if (message_view.transport_data_buffer()) {
        size_t num_platform_handles;
        const void* platform_handle_table;
        TransportData::GetPlatformHandleTable(
            message_view.transport_data_buffer(), &num_platform_handles,
            &platform_handle_table);

        if (num_platform_handles > 0) {
          platform_handles = GetReadPlatformHandles(num_platform_handles,
                                                    platform_handle_table);
          if (!platform_handles) {
            LOG(ERROR) << "Invalid number of platform handles received";
            CallOnError(Delegate::ERROR_READ_BAD_MESSAGE);
            *stop_dispatching = true;
            return;  // |this| may have been destroyed in |CallOnError()|.
          }
        }
      }

      // TODO(vtl): In the case that we aren't expecting any platform handles,
      // for the POSIX implementation, we should confirm that none are stored.
      if (delegate_) {
        DCHECK(!calling_delegate_);
        calling_delegate_ = true;
        delegate_->OnReadMessage(message_view, std::move(platform_handles));
        calling_delegate_ = false;
      }
    }

    *did_dispatch_message = true;

    // Update our state.
    read_buffer_start += message_size;
    remaining_bytes -= message_size;
  }

  if (read_buffer_start > 0) {
    // Move data back to start.
    read_buffer_->num_valid_bytes_ = remaining_bytes;
    if (read_buffer_->num_valid_bytes_ > 0) {
      memmove(&read_buffer_->buffer_[0],
              &read_buffer_->buffer_[read_buffer_start], remaining_bytes);
    }
    read_buffer_start = 0;
  }
}

void RawChannel::UpdateWriteBuffer(size_t platform_handles_written,
                                   size_t bytes_written) {
  write_buffer_->platform_handles_offset_ += platform_handles_written;
  write_buffer_->data_offset_ += bytes_written;

  MessageInTransit* message = write_buffer_->message_queue_.PeekMessage();
  if (write_buffer_->data_offset_ >= message->total_size()) {
    // Complete write.
    CHECK_EQ(write_buffer_->data_offset_, message->total_size());
    write_buffer_->message_queue_.DiscardMessage();
    write_buffer_->platform_handles_offset_ = 0;
    write_buffer_->data_offset_ = 0;
  }
}

void RawChannel::CallOnReadCompleted(IOResult io_result, size_t bytes_read) {
  base::AutoLock locker(read_lock_);
  OnReadCompletedNoLock(io_result, bytes_read);
}

void RawChannel::WillDestroyCurrentMessageLoop() {
  base::AutoLock locker(read_lock_);
  OnReadCompletedNoLock(IO_FAILED_SHUTDOWN, 0);
}

}  // namespace edk
}  // namespace mojo