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

(We weren't do so properly in the "too many handles" case.)

R=yzshen@chromium.org

Review URL: https://codereview.chromium.org/463883004

Cr-Commit-Position: refs/heads/master@{#289310}
git-svn-id: svn://svn.chromium.org/chrome/trunk/src@289310 0039d316-1c4b-4281-b951-d872f2087c98

mojo/system/raw_channel_posix.cc

@@ -63,6 +63,9 @@ class RawChannelPosix : public RawChannel,
   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();
 
@@ -173,55 +176,11 @@ 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);
+  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;
 }
 
@@ -445,6 +404,52 @@ void RawChannelPosix::OnFileCanWriteWithoutBlocking(int fd) {
     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());