Add Socket::ReadIfReady()

net/socket/socket_posix.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 "net/socket/socket_posix.h"
 
 #include <errno.h>
 #include <netinet/in.h>
 #include <sys/socket.h>
 #include <utility>
@@ skipping 44 matching lines @@
     }
   }
 }
 
 }  // namespace
 
 SocketPosix::SocketPosix()
     : socket_fd_(kInvalidSocket),
       read_buf_len_(0),
       write_buf_len_(0),
-      waiting_connect_(false) {}
+      waiting_connect_(false),
+      weak_factory_(this) {}
 
 SocketPosix::~SocketPosix() {
   Close();
 }
 
 int SocketPosix::Open(int address_family) {
   DCHECK(thread_checker_.CalledOnValidThread());
   DCHECK_EQ(kInvalidSocket, socket_fd_);
   DCHECK(address_family == AF_INET ||
          address_family == AF_INET6 ||
@@ skipping 167 matching lines @@
     return false;
   if (errno != EAGAIN && errno != EWOULDBLOCK)
     return false;
 
   return true;
 }
 
 int SocketPosix::Read(IOBuffer* buf,
                       int buf_len,
                       const CompletionCallback& callback) {
+  int rv = ReadIfReady(buf, buf_len, base::Bind(&SocketPosix::RetryRead,
+                                                weak_factory_.GetWeakPtr()));

[davidben, 2017/02/10 23:33:48]

Do you need a WeakPtr here? Since the callback won't get called after the object
is destroyed, I would think base::Unretained(this) would work fine.

[xunjieli, 2017/02/13 20:28:18]

Done. You are right. I confused this one with the one where you stash callback
onto a task runner.

+  if (rv == ERR_IO_PENDING) {
+    read_buf_ = buf;
+    read_buf_len_ = buf_len;
+    read_callback_ = callback;
+  }
+  return rv;
+}
+
+int SocketPosix::ReadIfReady(IOBuffer* buf,
+                             int buf_len,
+                             const CompletionCallback& callback) {
   DCHECK(thread_checker_.CalledOnValidThread());
   DCHECK_NE(kInvalidSocket, socket_fd_);
   DCHECK(!waiting_connect_);
-  CHECK(read_callback_.is_null());
+  CHECK(read_if_ready_callback_.is_null());
   // Synchronous operation not supported
   DCHECK(!callback.is_null());
   DCHECK_LT(0, buf_len);
 
   int rv = DoRead(buf, buf_len);
   if (rv != ERR_IO_PENDING)
     return rv;
 
   if (!base::MessageLoopForIO::current()->WatchFileDescriptor(
           socket_fd_, true, base::MessageLoopForIO::WATCH_READ,
           &read_socket_watcher_, this)) {
     PLOG(ERROR) << "WatchFileDescriptor failed on read, errno " << errno;
     return MapSystemError(errno);
   }
 
-  read_buf_ = buf;
-  read_buf_len_ = buf_len;
-  read_callback_ = callback;
+  read_if_ready_callback_ = callback;
   return ERR_IO_PENDING;
 }
 
 int SocketPosix::Write(IOBuffer* buf,
                        int buf_len,
                        const CompletionCallback& callback) {
   DCHECK(thread_checker_.CalledOnValidThread());
   DCHECK_NE(kInvalidSocket, socket_fd_);
   DCHECK(!waiting_connect_);
   CHECK(write_callback_.is_null());
@@ skipping 80 matching lines @@
   }
 }
 
 void SocketPosix::DetachFromThread() {
   thread_checker_.DetachFromThread();
 }
 
 void SocketPosix::OnFileCanReadWithoutBlocking(int fd) {
   TRACE_EVENT0(kNetTracingCategory,
                "SocketPosix::OnFileCanReadWithoutBlocking");
-  DCHECK(!accept_callback_.is_null() || !read_callback_.is_null());
+  DCHECK(!accept_callback_.is_null() || !read_if_ready_callback_.is_null());
   if (!accept_callback_.is_null()) {
     AcceptCompleted();
-  } else {  // !read_callback_.is_null()
+  } else {  // !read_if_ready_callback_.is_null()
     ReadCompleted();
   }
 }
 
 void SocketPosix::OnFileCanWriteWithoutBlocking(int fd) {
   DCHECK(!write_callback_.is_null());
   if (waiting_connect_) {
     ConnectCompleted();
   } else {
     WriteCompleted();
@@ skipping 54 matching lines @@
   DCHECK(ok);
   waiting_connect_ = false;
   base::ResetAndReturn(&write_callback_).Run(rv);
 }
 
 int SocketPosix::DoRead(IOBuffer* buf, int buf_len) {
   int rv = HANDLE_EINTR(read(socket_fd_, buf->data(), buf_len));
   return rv >= 0 ? rv : MapSystemError(errno);
 }
 
-void SocketPosix::ReadCompleted() {
-  int rv = DoRead(read_buf_.get(), read_buf_len_);
-  if (rv == ERR_IO_PENDING)
-    return;
+void SocketPosix::RetryRead(int rv) {
+  DCHECK(read_callback_);
+  DCHECK(read_buf_);
+  DCHECK_LT(0, read_buf_len_);
 
-  bool ok = read_socket_watcher_.StopWatchingFileDescriptor();
-  DCHECK(ok);
-  read_buf_ = NULL;
+  if (rv == OK) {
+    rv = ReadIfReady(
+        read_buf_.get(), read_buf_len_,
+        base::Bind(&SocketPosix::RetryRead, weak_factory_.GetWeakPtr()));
+    if (rv == ERR_IO_PENDING)
+      return;
+  }
+  read_buf_ = nullptr;
   read_buf_len_ = 0;
   base::ResetAndReturn(&read_callback_).Run(rv);
 }
 
+void SocketPosix::ReadCompleted() {
+  DCHECK(read_if_ready_callback_);
+
+  bool ok = read_socket_watcher_.StopWatchingFileDescriptor();
+  DCHECK(ok);
+  base::ResetAndReturn(&read_if_ready_callback_).Run(OK);
+}
+
 int SocketPosix::DoWrite(IOBuffer* buf, int buf_len) {
 #if defined(OS_LINUX) || defined(OS_ANDROID)
   // Disable SIGPIPE for this write. Although Chromium globally disables
   // SIGPIPE, the net stack may be used in other consumers which do not do
   // this. MSG_NOSIGNAL is a Linux-only API. On OS X, this is a setsockopt on
   // socket creation.
   int rv = HANDLE_EINTR(send(socket_fd_, buf->data(), buf_len, MSG_NOSIGNAL));
 #else
   int rv = HANDLE_EINTR(write(socket_fd_, buf->data(), buf_len));
 #endif
@@ skipping 35 matching lines @@
     write_buf_ = NULL;
     write_buf_len_ = 0;
     write_callback_.Reset();
   }
 
   waiting_connect_ = false;
   peer_address_.reset();
 }
 
 }  // namespace net