RawChannel refactoring

mojo/system/raw_channel_posix.cc

 // 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.
 
 #include "mojo/system/raw_channel.h"
 
 #include <errno.h>
-#include <string.h>
 #include <sys/uio.h>
 #include <unistd.h>
 
-#include <algorithm>
-#include <deque>
-#include <vector>
-
 #include "base/basictypes.h"
 #include "base/bind.h"
 #include "base/compiler_specific.h"
 #include "base/location.h"
 #include "base/logging.h"
 #include "base/memory/scoped_ptr.h"
 #include "base/memory/weak_ptr.h"
 #include "base/message_loop/message_loop.h"
 #include "base/posix/eintr_wrapper.h"
-#include "base/stl_util.h"
 #include "base/synchronization/lock.h"
 #include "mojo/system/embedder/platform_handle.h"
-#include "mojo/system/message_in_transit.h"
 
 namespace mojo {
 namespace system {
 
 namespace {
 
-const size_t kReadSize = 4096;
-
 class RawChannelPosix : public RawChannel,
                         public base::MessageLoopForIO::Watcher {
  public:
   RawChannelPosix(embedder::ScopedPlatformHandle handle,
                   Delegate* delegate,
                   base::MessageLoopForIO* message_loop_for_io);
   virtual ~RawChannelPosix();
 
+ private:
   // |RawChannel| implementation:
-  virtual bool Init() OVERRIDE;
-  virtual void Shutdown() OVERRIDE;
-  virtual bool WriteMessage(scoped_ptr<MessageInTransit> message) OVERRIDE;
+  virtual IOResult Read(size_t* bytes_read) OVERRIDE;
+  virtual IOResult ScheduleRead() OVERRIDE;
+  virtual IOResult WriteNoLock(size_t* bytes_written) OVERRIDE;
+  virtual IOResult ScheduleWriteNoLock() OVERRIDE;
+  virtual bool OnInit() OVERRIDE;
+  virtual void OnShutdownNoLock(
+      scoped_ptr<ReadBuffer> read_buffer,
+      scoped_ptr<WriteBuffer> write_buffer) OVERRIDE;
 
- private:
   // |base::MessageLoopForIO::Watcher| implementation:
   virtual void OnFileCanReadWithoutBlocking(int fd) OVERRIDE;
   virtual void OnFileCanWriteWithoutBlocking(int fd) OVERRIDE;
 
   // Watches for |fd_| to become writable. Must be called on the I/O thread.
   void WaitToWrite();
 
-  // Calls |delegate()->OnFatalError(fatal_error)|. Must be called on the I/O
-  // thread WITHOUT |write_lock_| held.
-  void CallOnFatalError(Delegate::FatalError fatal_error);
-
-  // Writes the message at the front of |write_message_queue_|, starting at
-  // |write_message_offset_|. It removes and destroys if the write completes and
-  // otherwise updates |write_message_offset_|. Returns true on success. Must be
-  // called under |write_lock_|.
-  bool WriteFrontMessageNoLock();
-
-  // Cancels all pending writes and destroys the contents of
-  // |write_message_queue_|. Should only be called if |write_stopped_| is false;
-  // sets |write_stopped_| to true. Must be called under |write_lock_|.
-  void CancelPendingWritesNoLock();
-
   embedder::ScopedPlatformHandle fd_;
 
-  // Only used on the I/O thread:
+  // The following members are only used on the I/O thread:
   scoped_ptr<base::MessageLoopForIO::FileDescriptorWatcher> read_watcher_;
   scoped_ptr<base::MessageLoopForIO::FileDescriptorWatcher> write_watcher_;
 
-  // We store data from |read()|s in |read_buffer_|. The start of |read_buffer_|
-  // is always aligned with a message boundary (we will copy memory to ensure
-  // this), but |read_buffer_| may be larger than the actual number of bytes we
-  // have.
-  std::vector<char> read_buffer_;
-  size_t read_buffer_num_valid_bytes_;
+  bool pending_read_;
 
-  base::Lock write_lock_;  // Protects the following members.
-  bool write_stopped_;
-  // TODO(vtl): When C++11 is available, switch this to a deque of
-  // |scoped_ptr|/|unique_ptr|s.
-  std::deque<MessageInTransit*> write_message_queue_;
-  size_t write_message_offset_;
+  // The following members are used on multiple threads and protected by
+  // |write_lock()|:
+  bool pending_write_;
+
   // This is used for posting tasks from write threads to the I/O thread. It
   // must only be accessed under |write_lock_|. The weak pointers it produces
   // are only used/invalidated on the I/O thread.
   base::WeakPtrFactory<RawChannelPosix> weak_ptr_factory_;
 
   DISALLOW_COPY_AND_ASSIGN(RawChannelPosix);
 };
 
 RawChannelPosix::RawChannelPosix(embedder::ScopedPlatformHandle handle,
                                  Delegate* delegate,
                                  base::MessageLoopForIO* message_loop_for_io)
     : RawChannel(delegate, message_loop_for_io),
       fd_(handle.Pass()),
-      read_buffer_num_valid_bytes_(0),
-      write_stopped_(false),
-      write_message_offset_(0),
+      pending_read_(false),
+      pending_write_(false),
       weak_ptr_factory_(this) {
-  CHECK_EQ(RawChannel::message_loop_for_io()->type(),
-           base::MessageLoop::TYPE_IO);
   DCHECK(fd_.is_valid());
 }
 
 RawChannelPosix::~RawChannelPosix() {
-  DCHECK(write_stopped_);
-  DCHECK(!fd_.is_valid());
+  DCHECK(!pending_read_);
+  DCHECK(!pending_write_);
 
-  // No need to take the |write_lock_| here -- if there are still weak pointers
+  // No need to take the |write_lock()| here -- if there are still weak pointers
   // outstanding, then we're hosed anyway (since we wouldn't be able to
   // invalidate them cleanly, since we might not be on the I/O thread).
   DCHECK(!weak_ptr_factory_.HasWeakPtrs());
 
   // These must have been shut down/destroyed on the I/O thread.
   DCHECK(!read_watcher_.get());
   DCHECK(!write_watcher_.get());
 }
 
-bool RawChannelPosix::Init() {
+RawChannel::IOResult RawChannelPosix::Read(size_t* bytes_read) {
+  DCHECK_EQ(base::MessageLoop::current(), message_loop_for_io());
+  DCHECK(!pending_read_);
+
+  char* buffer = NULL;
+  size_t bytes_to_read = 0;
+  read_buffer()->GetBuffer(&buffer, &bytes_to_read);
+
+  ssize_t read_result = HANDLE_EINTR(read(fd_.get().fd, buffer, bytes_to_read));
+
+  if (read_result >= 0) {
+    *bytes_read = static_cast<size_t>(read_result);
+    return IO_SUCCEEDED;
+  } else if (errno != EAGAIN && errno != EWOULDBLOCK) {

[viettrungluu, 2014/02/27 05:22:15]

nit: no |else| needed here

[yzshen1, 2014/02/27 06:56:18]

Done.

+    PLOG(ERROR) << "read";
+
+    // Make sure that |OnFileCanReadWithoutBlocking()| won't be called again.
+    read_watcher_.reset();
+
+    return IO_FAILED;
+  }
+
+  return ScheduleRead();
+}
+
+RawChannel::IOResult RawChannelPosix::ScheduleRead() {
+  DCHECK_EQ(base::MessageLoop::current(), message_loop_for_io());
+  DCHECK(!pending_read_);
+
+  pending_read_ = true;
+
+  return IO_PENDING;
+}
+
+RawChannel::IOResult RawChannelPosix::WriteNoLock(size_t* bytes_written) {
+  write_lock().AssertAcquired();
+
+  DCHECK(!pending_write_);
+
+  WriteBuffer::BufferVector buffers;
+  write_buffer_no_lock()->GetBuffers(&buffers);
+  DCHECK(!buffers.empty());
+
+  ssize_t write_result = -1;
+  if (buffers.size() == 1) {
+    write_result = HANDLE_EINTR(
+        write(fd_.get().fd, buffers[0].first, buffers[0].second));
+  } else {
+    // Note that using |writev()| is measurably slower than using |write()| --
+    // at least in a microbenchmark -- but much faster than using multiple
+    // |write()|s.
+    iovec* iov = new iovec[buffers.size()];

[viettrungluu, 2014/02/27 05:22:15]

I think it's overkill to do a heap allocation for this. I'd just allocate N
(where N is some reasonable size, say 10), and then iterate for i <
std::min(buffers.size(), N).

To avoid copying vectors around and such, it might be reasonable to change
WriteBuffer's API to instead be:

const WriteBuffer::Buffer* GetBuffer(size_t index);

(returning null if index > number of segments).

To make |GetBuffer()| work, you'd just store an array/vector of
|WriteBuffer::Buffer|s (currently with exactly two entries) in the object.

[yzshen1, 2014/02/27 06:56:18]

Done.
Storing a vector of |WriteBuffer::Buffer| requires an explicit action to
"update" it and maintain its consistency with other data members. For now,
however, WriteBuffer lets RawChannel directly access its members so the burden
of maintaining consistency will be on RawChannel. Besides, IMO currently it is
not very expensive to compute the vector for every call of GetBuffers() (at most
two elements). If you don't mind, I would prefer to do this optimization later,
when we want to encapsulate more logic in WriteBuffer, or when GetBuffers()
becomes more expensive.
What do you think?

[viettrungluu, 2014/02/27 07:36:35]

Isn't consistency guaranteed by the fact that access is all done under
write_lock()?

(Consistency is already implied, since the WriteBuffer isn't allowed to be
changed until the write completes.)

[yzshen1, 2014/02/27 16:19:33]

I mean:

write_buffer_->message_queue_->pop_front();
write_buffer_->offset_ = 0;
[write_buffer_->]UpdateBuffersVector(); // I will need this explicit action to
update the stored std::vector<Buffer>, so that it is consistent with
message_queue_/offset_. (Because they contain duplicated information.)

The current usage of GetBuffers() (called only once per WriteNoLock()) and the
cost of generating the vector (at most two elements), makes me think that it
doesn't worth the effort. But if later we want to encapsulate more logic in
WriteBuffer so that it maintains consistency across its own members; or
GetBuffers() becomes more expensive, I agree that stored std::vector<Buffer> is
a way of optimization.

[viettrungluu, 2014/02/27 17:21:41]

Okay, LGTM then.

+    for (size_t i = 0; i < buffers.size(); ++i) {
+      iov[i].iov_base = const_cast<char*>(buffers[i].first);
+      iov[i].iov_len = buffers[i].second;
+    }
+
+    write_result = HANDLE_EINTR(
+        writev(fd_.get().fd, iov, buffers.size()));
+    delete [] iov;
+  }
+
+  if (write_result >= 0) {
+    *bytes_written = static_cast<size_t>(write_result);
+    return IO_SUCCEEDED;
+  } else if (errno != EAGAIN && errno != EWOULDBLOCK) {

[viettrungluu, 2014/02/27 05:22:15]

no else

[yzshen1, 2014/02/27 06:56:18]

Done.

+    PLOG(ERROR) << "write of size "
+                << write_buffer_no_lock()->GetTotalBytesToWrite();
+    return IO_FAILED;
+  }
+
+  return ScheduleWriteNoLock();
+}
+
+RawChannel::IOResult RawChannelPosix::ScheduleWriteNoLock() {
+  write_lock().AssertAcquired();
+
+  DCHECK(!pending_write_);
+
+  // Set up to wait for the FD to become writable.
+  // If we're not on the I/O thread, we have to post a task to do this.
+  if (base::MessageLoop::current() != message_loop_for_io()) {
+    message_loop_for_io()->PostTask(
+        FROM_HERE,
+        base::Bind(&RawChannelPosix::WaitToWrite,
+                   weak_ptr_factory_.GetWeakPtr()));
+    pending_write_ = true;
+    return IO_PENDING;
+  }
+
+  if (message_loop_for_io()->WatchFileDescriptor(
+      fd_.get().fd, false, base::MessageLoopForIO::WATCH_WRITE,
+      write_watcher_.get(), this)) {
+    pending_write_ = true;
+    return IO_PENDING;
+  }
+
+  return IO_FAILED;
+}
+
+bool RawChannelPosix::OnInit() {
   DCHECK_EQ(base::MessageLoop::current(), message_loop_for_io());
 
   DCHECK(!read_watcher_.get());
   read_watcher_.reset(new base::MessageLoopForIO::FileDescriptorWatcher());
   DCHECK(!write_watcher_.get());
   write_watcher_.reset(new base::MessageLoopForIO::FileDescriptorWatcher());
 
-  // No need to take the lock. No one should be using us yet.
-  DCHECK(write_message_queue_.empty());
-
   if (!message_loop_for_io()->WatchFileDescriptor(fd_.get().fd, true,
           base::MessageLoopForIO::WATCH_READ, read_watcher_.get(), this)) {
     // TODO(vtl): I'm not sure |WatchFileDescriptor()| actually fails cleanly
     // (in the sense of returning the message loop's state to what it was before
     // it was called).
     read_watcher_.reset();
     write_watcher_.reset();
     return false;
   }
 
   return true;
 }
 
-void RawChannelPosix::Shutdown() {
-  DCHECK_EQ(base::MessageLoop::current(), message_loop_for_io());
-
-  base::AutoLock locker(write_lock_);
-  if (!write_stopped_)
-    CancelPendingWritesNoLock();
+void RawChannelPosix::OnShutdownNoLock(
+    scoped_ptr<ReadBuffer> /*read_buffer*/,
+    scoped_ptr<WriteBuffer> /*write_buffer*/) {
+  DCHECK_EQ(base::MessageLoop::current(), message_loop_for_io());
+  write_lock().AssertAcquired();
 
   read_watcher_.reset();  // This will stop watching (if necessary).
   write_watcher_.reset();  // This will stop watching (if necessary).
 
+  pending_read_ = false;
+  pending_write_ = false;
+
   DCHECK(fd_.is_valid());
   fd_.reset();
 
   weak_ptr_factory_.InvalidateWeakPtrs();
 }
 
-// Reminder: This must be thread-safe, and takes ownership of |message|.
-bool RawChannelPosix::WriteMessage(scoped_ptr<MessageInTransit> message) {
-  base::AutoLock locker(write_lock_);
-  if (write_stopped_)
-    return false;
-
-  if (!write_message_queue_.empty()) {
-    write_message_queue_.push_back(message.release());
-    return true;
-  }
-
-  write_message_queue_.push_front(message.release());
-  DCHECK_EQ(write_message_offset_, 0u);
-  bool result = WriteFrontMessageNoLock();
-  DCHECK(result || write_message_queue_.empty());
-
-  if (!result) {
-    // Even if we're on the I/O thread, don't call |OnFatalError()| in the
-    // nested context.
-    message_loop_for_io()->PostTask(
-        FROM_HERE,
-        base::Bind(&RawChannelPosix::CallOnFatalError,
-                   weak_ptr_factory_.GetWeakPtr(),
-                   Delegate::FATAL_ERROR_FAILED_WRITE));
-  } else if (!write_message_queue_.empty()) {
-    // Set up to wait for the FD to become writable. If we're not on the I/O
-    // thread, we have to post a task to do this.
-    if (base::MessageLoop::current() == message_loop_for_io()) {
-      WaitToWrite();
-    } else {
-      message_loop_for_io()->PostTask(
-          FROM_HERE,
-          base::Bind(&RawChannelPosix::WaitToWrite,
-                     weak_ptr_factory_.GetWeakPtr()));
-    }
-  }
-
-  return result;
-}
-
 void RawChannelPosix::OnFileCanReadWithoutBlocking(int fd) {
   DCHECK_EQ(fd, fd_.get().fd);
   DCHECK_EQ(base::MessageLoop::current(), message_loop_for_io());
 
-  bool did_dispatch_message = 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;
-  for (;;) {
-    if (read_buffer_.size() - (read_buffer_start + 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_.size(), kReadSize);
-      while (new_size <
-                 read_buffer_start + read_buffer_num_valid_bytes_ + kReadSize)
-        new_size *= 2;
-
-      // TODO(vtl): It's suboptimal to zero out the fresh memory.
-      read_buffer_.resize(new_size, 0);
-    }
-
-    ssize_t bytes_read = HANDLE_EINTR(
-        read(fd_.get().fd,
-             &read_buffer_[read_buffer_start + read_buffer_num_valid_bytes_],
-             kReadSize));
-    if (bytes_read < 0) {
-      if (errno != EAGAIN && errno != EWOULDBLOCK) {
-        PLOG(ERROR) << "read";
-
-        // Make sure that |OnFileCanReadWithoutBlocking()| won't be called
-        // again.
-        read_watcher_.reset();
-
-        CallOnFatalError(Delegate::FATAL_ERROR_FAILED_READ);
-        return;
-      }
-
-      break;
-    }
-
-    read_buffer_num_valid_bytes_ += static_cast<size_t>(bytes_read);
-
-    // Dispatch all the messages that we can.
-    size_t message_size;
-    // Note that we rely on short-circuit evaluation here:
-    //   - |read_buffer_start| may be an invalid index into |read_buffer_| if
-    //     |read_buffer_num_valid_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).
-    while (read_buffer_num_valid_bytes_ > 0 &&
-           MessageInTransit::GetNextMessageSize(
-               &read_buffer_[read_buffer_start], read_buffer_num_valid_bytes_,
-               &message_size) &&
-           read_buffer_num_valid_bytes_ >= message_size) {
-      // TODO(vtl): FIXME -- replace "unowned buffer" |MessageInTransit|s with
-      // some sort of "view" abstraction.
-      MessageInTransit message(MessageInTransit::UNOWNED_BUFFER, message_size,
-                               &read_buffer_[read_buffer_start]);
-      DCHECK_EQ(message.total_size(), message_size);
-
-      // Dispatch the message.
-      delegate()->OnReadMessage(message);
-      if (!read_watcher_.get()) {
-        // |Shutdown()| was called in |OnReadMessage()|.
-        // TODO(vtl): Add test for this case.
-        return;
-      }
-      did_dispatch_message = true;
-
-      // Update our state.
-      read_buffer_start += message_size;
-      read_buffer_num_valid_bytes_ -= message_size;
-    }
-
-    // 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.)
-    if (did_dispatch_message)
-      break;
-
-    // If we didn't max out |kReadSize|, stop reading for now.
-    if (static_cast<size_t>(bytes_read) < kReadSize)
-      break;
-
-    // Else try to read some more....
-  }
-
-  // Move data back to start.
-  if (read_buffer_start > 0) {
-    if (read_buffer_num_valid_bytes_ > 0) {
-      memmove(&read_buffer_[0], &read_buffer_[read_buffer_start],
-              read_buffer_num_valid_bytes_);
-    }
-    read_buffer_start = 0;
-  }
+  if (!pending_read_) {
+    NOTREACHED();
+    return;
+  }
+
+  pending_read_ = false;
+  size_t bytes_read = 0;
+  IOResult result = Read(&bytes_read);
+  if (result != IO_PENDING)
+    OnReadCompleted(result == IO_SUCCEEDED, bytes_read);
+
+  // On failure, |read_watcher_| must have been reset; on success,
+  // we assume that |OnReadCompleted()| always schedules another read.
+  // Otherwise, we could end up spinning -- getting
+  // |OnFileCanReadWithoutBlocking()| again and again but not doing any actual
+  // read.
+  // TODO(yzshen): An alternative is to stop watching if RawChannel doesn't
+  // schedule a new read. But that code won't be reached under the current
+  // RawChannel implementation.
+  DCHECK(!read_watcher_.get() || pending_read_);
 }
 
 void RawChannelPosix::OnFileCanWriteWithoutBlocking(int fd) {
   DCHECK_EQ(fd, fd_.get().fd);
   DCHECK_EQ(base::MessageLoop::current(), message_loop_for_io());
 
-  bool did_fail = false;
+  IOResult result = IO_FAILED;
+  size_t bytes_written = 0;
   {
-    base::AutoLock locker(write_lock_);
-    DCHECK_EQ(write_stopped_, write_message_queue_.empty());
-
-    if (write_stopped_) {
-      write_watcher_.reset();
-      return;
+    base::AutoLock locker(write_lock());
+
+    DCHECK(pending_write_);
+
+    pending_write_ = false;
+    result = WriteNoLock(&bytes_written);
+  }
+
+  if (result != IO_PENDING)
+    OnWriteCompleted(result == IO_SUCCEEDED, bytes_written);
+}
+
+void RawChannelPosix::WaitToWrite() {
+  DCHECK_EQ(base::MessageLoop::current(), message_loop_for_io());
+
+  DCHECK(write_watcher_.get());
+
+  if (!message_loop_for_io()->WatchFileDescriptor(
+          fd_.get().fd, false, base::MessageLoopForIO::WATCH_WRITE,
+          write_watcher_.get(), this)) {
+    {
+      base::AutoLock locker(write_lock());
+
+      DCHECK(pending_write_);
+      pending_write_ = false;
     }
-
-    bool result = WriteFrontMessageNoLock();
-    DCHECK(result || write_message_queue_.empty());
-
-    if (!result) {
-      did_fail = true;
-      write_watcher_.reset();
-    } else if (!write_message_queue_.empty()) {
-      WaitToWrite();
-    }
-  }
-  if (did_fail)
-    CallOnFatalError(Delegate::FATAL_ERROR_FAILED_WRITE);
-}
-
-void RawChannelPosix::WaitToWrite() {
-  DCHECK_EQ(base::MessageLoop::current(), message_loop_for_io());
-
-  DCHECK(write_watcher_.get());
-  bool result = message_loop_for_io()->WatchFileDescriptor(
-      fd_.get().fd, false, base::MessageLoopForIO::WATCH_WRITE,
-      write_watcher_.get(), this);
-  DCHECK(result);
-}
-
-void RawChannelPosix::CallOnFatalError(Delegate::FatalError fatal_error) {
-  DCHECK_EQ(base::MessageLoop::current(), message_loop_for_io());
-  // TODO(vtl): Add a "write_lock_.AssertNotAcquired()"?
-  delegate()->OnFatalError(fatal_error);
-}
-
-// TODO(vtl): This will collide with yzshen's work. This is just a hacky,
-// minimally-invasive function that does what I want (write/resume writing a
-// |MessageInTransit| that may consist of more than one buffer).
-ssize_t WriteMessageInTransit(int fd,
-                              MessageInTransit* message,
-                              size_t offset,
-                              size_t* bytes_to_write) {
-  *bytes_to_write = message->total_size() - offset;
-  if (!message->secondary_buffer_size()) {
-    // Only write from the main buffer.
-    DCHECK_LT(offset, message->main_buffer_size());
-    DCHECK_LE(*bytes_to_write, message->main_buffer_size());
-    return HANDLE_EINTR(
-        write(fd,
-              static_cast<const char*>(message->main_buffer()) + offset,
-              *bytes_to_write));
-  }
-  if (offset >= message->main_buffer_size()) {
-    // Only write from the secondary buffer.
-    DCHECK_LT(offset - message->main_buffer_size(),
-              message->secondary_buffer_size());
-    DCHECK_LE(*bytes_to_write, message->secondary_buffer_size());
-    return HANDLE_EINTR(
-        write(fd,
-              static_cast<const char*>(message->secondary_buffer()) +
-                  (offset - message->main_buffer_size()),
-              *bytes_to_write));
-  }
-  // Write from both buffers. (Note that using |writev()| is measurably slower
-  // than using |write()| -- at least in a microbenchmark -- but much faster
-  // than using two |write()|s.)
-  DCHECK_EQ(*bytes_to_write, message->main_buffer_size() - offset +
-                                 message->secondary_buffer_size());
-  struct iovec iov[2] = {
-    { const_cast<char*>(
-          static_cast<const char*>(message->main_buffer()) + offset),
-      message->main_buffer_size() - offset },
-    { const_cast<void*>(message->secondary_buffer()),
-      message->secondary_buffer_size() }
-  };
-  return HANDLE_EINTR(writev(fd, iov, 2));
-}
-
-bool RawChannelPosix::WriteFrontMessageNoLock() {
-  write_lock_.AssertAcquired();
-
-  DCHECK(!write_stopped_);
-  DCHECK(!write_message_queue_.empty());
-
-  MessageInTransit* message = write_message_queue_.front();
-  DCHECK_LT(write_message_offset_, message->total_size());
-  size_t bytes_to_write;
-  ssize_t bytes_written = WriteMessageInTransit(fd_.get().fd,
-                                                message,
-                                                write_message_offset_,
-                                                &bytes_to_write);
-  if (bytes_written < 0) {
-    if (errno != EAGAIN && errno != EWOULDBLOCK) {
-      PLOG(ERROR) << "write of size " << bytes_to_write;
-      CancelPendingWritesNoLock();
-      return false;
-    }
-
-    // We simply failed to write since we'd block. The logic is the same as if
-    // we got a partial write.
-    bytes_written = 0;
-  }
-
-  DCHECK_GE(bytes_written, 0);
-  if (static_cast<size_t>(bytes_written) < bytes_to_write) {
-    // Partial (or no) write.
-    write_message_offset_ += static_cast<size_t>(bytes_written);
-  } else {
-    // Complete write.
-    DCHECK_EQ(static_cast<size_t>(bytes_written), bytes_to_write);
-    write_message_queue_.pop_front();
-    delete message;
-    write_message_offset_ = 0;
-  }
-
-  return true;
-}
-
-void RawChannelPosix::CancelPendingWritesNoLock() {
-  write_lock_.AssertAcquired();
-  DCHECK(!write_stopped_);
-
-  write_stopped_ = true;
-  STLDeleteElements(&write_message_queue_);
+    OnWriteCompleted(false, 0);
+  }
 }
 
 }  // namespace
 
 // -----------------------------------------------------------------------------
 
 // Static factory method declared in raw_channel.h.
 // static
 RawChannel* RawChannel::Create(embedder::ScopedPlatformHandle handle,
                                Delegate* delegate,
                                base::MessageLoopForIO* message_loop_for_io) {
   return new RawChannelPosix(handle.Pass(), delegate, message_loop_for_io);
 }
 
 }  // namespace system
 }  // namespace mojo