Mojo: Properly cancel further read on all read failures in RawChannelPosix.

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 <sys/uio.h>
 #include <unistd.h>
 
@@ skipping 45 matching lines @@
                                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;
 
   // |base::MessageLoopForIO::Watcher| implementation:
   virtual void OnFileCanReadWithoutBlocking(int fd) OVERRIDE;
   virtual void OnFileCanWriteWithoutBlocking(int fd) OVERRIDE;
 
+  // Implements most of |Read()| (except for a bit of clean-up):
+  IOResult ReadImpl(size_t* bytes_read);
+
   // Watches for |fd_| to become writable. Must be called on the I/O thread.
   void WaitToWrite();
 
   embedder::ScopedPlatformHandle fd_;
 
   // 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_;
 
   bool pending_read_;
@@ skipping 90 matching lines @@
     return true;
   }
 
   return RawChannel::OnReadMessageForRawChannel(message_view);
 }
 
 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);
-
-  size_t old_num_platform_handles = read_platform_handles_.size();
-  ssize_t read_result = embedder::PlatformChannelRecvmsg(
-      fd_.get(), buffer, bytes_to_read, &read_platform_handles_);
-  if (read_platform_handles_.size() > old_num_platform_handles) {
-    DCHECK_LE(read_platform_handles_.size() - old_num_platform_handles,
-              embedder::kPlatformChannelMaxNumHandles);
-
-    // We should never accumulate more than |TransportData::kMaxPlatformHandles
-    // + embedder::kPlatformChannelMaxNumHandles| handles. (The latter part is
-    // possible because we could have accumulated all the handles for a message,
-    // then received the message data plus the first set of handles for the next
-    // message in the subsequent |recvmsg()|.)
-    if (read_platform_handles_.size() >
-        (TransportData::kMaxPlatformHandles +
-         embedder::kPlatformChannelMaxNumHandles)) {
-      LOG(ERROR) << "Received too many platform handles";
-      embedder::CloseAllPlatformHandles(&read_platform_handles_);
-      read_platform_handles_.clear();
-      return IO_FAILED_UNKNOWN;
-    }
-  }
-
-  if (read_result > 0) {
-    *bytes_read = static_cast<size_t>(read_result);
-    return IO_SUCCEEDED;
-  }
-
-  // |read_result == 0| means "end of file".
-  if (read_result == 0) {
+  IOResult rv = ReadImpl(bytes_read);

[yzshen1, 2014/08/12 21:21:17]

optional: I think for windows code, sometimes we use
do {
  ...
  break;
  ...
  return;  // On success
} while (false);
//error handling

+  if (rv != IO_SUCCEEDED && rv != IO_PENDING) {
     // Make sure that |OnFileCanReadWithoutBlocking()| won't be called again.
     read_watcher_.reset();
-    return IO_FAILED_SHUTDOWN;
   }
-
-  if (errno == EAGAIN || errno == EWOULDBLOCK)
-    return ScheduleRead();
-
-  IOResult rv = IO_FAILED_UNKNOWN;
-  if (errno == ECONNRESET)
-    rv = IO_FAILED_BROKEN;
-  else
-    PLOG(WARNING) << "recvmsg";
-
-  // Make sure that |OnFileCanReadWithoutBlocking()| won't be called again.
-  read_watcher_.reset();
   return rv;
 }
 
 RawChannel::IOResult RawChannelPosix::ScheduleRead() {
   DCHECK_EQ(base::MessageLoop::current(), message_loop_for_io());
   DCHECK(!pending_read_);
 
   pending_read_ = true;
 
   return IO_PENDING;
@@ skipping 203 matching lines @@
     DCHECK(pending_write_);
 
     pending_write_ = false;
     io_result = WriteNoLock(&platform_handles_written, &bytes_written);
   }
 
   if (io_result != IO_PENDING)
     OnWriteCompleted(io_result, platform_handles_written, bytes_written);
 }
 
+RawChannel::IOResult RawChannelPosix::ReadImpl(size_t* bytes_read) {
+  char* buffer = NULL;
+  size_t bytes_to_read = 0;
+  read_buffer()->GetBuffer(&buffer, &bytes_to_read);
+
+  size_t old_num_platform_handles = read_platform_handles_.size();
+  ssize_t read_result = embedder::PlatformChannelRecvmsg(
+      fd_.get(), buffer, bytes_to_read, &read_platform_handles_);
+  if (read_platform_handles_.size() > old_num_platform_handles) {
+    DCHECK_LE(read_platform_handles_.size() - old_num_platform_handles,
+              embedder::kPlatformChannelMaxNumHandles);
+
+    // We should never accumulate more than |TransportData::kMaxPlatformHandles
+    // + embedder::kPlatformChannelMaxNumHandles| handles. (The latter part is
+    // possible because we could have accumulated all the handles for a message,
+    // then received the message data plus the first set of handles for the next
+    // message in the subsequent |recvmsg()|.)
+    if (read_platform_handles_.size() >
+        (TransportData::kMaxPlatformHandles +
+         embedder::kPlatformChannelMaxNumHandles)) {
+      LOG(ERROR) << "Received too many platform handles";
+      embedder::CloseAllPlatformHandles(&read_platform_handles_);
+      read_platform_handles_.clear();
+      return IO_FAILED_UNKNOWN;
+    }
+  }
+
+  if (read_result > 0) {
+    *bytes_read = static_cast<size_t>(read_result);
+    return IO_SUCCEEDED;
+  }
+
+  // |read_result == 0| means "end of file".
+  if (read_result == 0)
+    return IO_FAILED_SHUTDOWN;
+
+  if (errno == EAGAIN || errno == EWOULDBLOCK)
+    return ScheduleRead();
+
+  if (errno == ECONNRESET)
+    return IO_FAILED_BROKEN;
+
+  PLOG(WARNING) << "recvmsg";
+  return IO_FAILED_UNKNOWN;
+}
+
 void RawChannelPosix::WaitToWrite() {
   DCHECK_EQ(base::MessageLoop::current(), message_loop_for_io());
 
   DCHECK(write_watcher_);
 
   if (!message_loop_for_io()->WatchFileDescriptor(
           fd_.get().fd,
           false,
           base::MessageLoopForIO::WATCH_WRITE,
           write_watcher_.get(),
@@ skipping 14 matching lines @@
 
 // Static factory method declared in raw_channel.h.
 // static
 scoped_ptr<RawChannel> RawChannel::Create(
     embedder::ScopedPlatformHandle handle) {
   return scoped_ptr<RawChannel>(new RawChannelPosix(handle.Pass()));
 }
 
 }  // namespace system
 }  // namespace mojo