Switch the TCP reads on Windows to use non-blocking/non-async I/O.

net/socket/tcp_client_socket_win.cc

 // Copyright (c) 2012 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/tcp_client_socket_win.h"
 
 #include <mstcpip.h>
 
 #include "base/basictypes.h"
 #include "base/compiler_specific.h"
@@ skipping 10 matching lines @@
 #include "net/base/network_change_notifier.h"
 #include "net/base/winsock_init.h"
 #include "net/base/winsock_util.h"
 #include "net/socket/socket_net_log_params.h"
 
 namespace net {
 
 namespace {
 
 const int kTCPKeepAliveSeconds = 45;
+const bool kUseNonBlockingSockets = true;

[slamm_google, 2012/08/30 17:47:25]

Refer to the related but that explains the reason for this approach?

[pmeenan, 2012/08/30 17:54:31]

Will be switching this to use an experiment once I figure out how to plumb that.

 
 bool SetSocketReceiveBufferSize(SOCKET socket, int32 size) {
   int rv = setsockopt(socket, SOL_SOCKET, SO_RCVBUF,
                       reinterpret_cast<const char*>(&size), sizeof(size));
   DCHECK(!rv) << "Could not set socket receive buffer size: " << GetLastError();
   return rv == 0;
 }
 
 bool SetSocketSendBufferSize(SOCKET socket, int32 size) {
   int rv = setsockopt(socket, SOL_SOCKET, SO_SNDBUF,
@@ skipping 255 matching lines @@
   write_watcher_.StartWatching(write_overlapped_.hEvent, &writer_);
 }
 
 void TCPClientSocketWin::Core::ReadDelegate::OnObjectSignaled(
     HANDLE object) {
   DCHECK_EQ(object, core_->read_overlapped_.hEvent);
   if (core_->socket_) {
     if (core_->socket_->waiting_connect()) {
       core_->socket_->DidCompleteConnect();
     } else {
-      core_->socket_->DidCompleteRead();
+      if (kUseNonBlockingSockets) {
+        core_->socket_->HandleReadSignalled();

[slamm_google, 2012/08/30 17:47:25]

Signalled -> Signaled

The dictionary has it both ways, however, this file already uses the latter.

[pmeenan, 2012/08/30 17:54:31]

Thanks.  Looked kind of strange.

+      } else {
+        core_->socket_->DidCompleteRead();
+      }
     }
   }
 
   core_->Release();
 }
 
 void TCPClientSocketWin::Core::WriteDelegate::OnObjectSignaled(
     HANDLE object) {
   DCHECK_EQ(object, core_->write_overlapped_.hEvent);
   if (core_->socket_)
@@ skipping 32 matching lines @@
   DCHECK_EQ(socket_, INVALID_SOCKET);
 
   int error = SetupSocket(socket);
   if (error)
     return MapSystemError(error);
 
   socket_ = socket;
   SetNonBlocking(socket_);
 
   core_ = new Core(this);
+  if (kUseNonBlockingSockets) {
+    WSAEventSelect(socket_, core_->read_overlapped_.hEvent,
+                   FD_READ | FD_CLOSE);
+  }
 
   current_address_index_ = 0;
   use_history_.set_was_ever_connected();
 
   return OK;
 }
 
 int TCPClientSocketWin::Bind(const IPEndPoint& address) {
   if (current_address_index_ >= 0 || bind_address_.get()) {
     // Cannot bind the socket if we are already connected or connecting.
@@ skipping 111 matching lines @@
         return ERR_INVALID_ARGUMENT;
       if (bind(socket_, storage.addr, storage.addr_len))
         return MapSystemError(errno);
     }
   }
 
   DCHECK(!core_);
   core_ = new Core(this);
   // WSAEventSelect sets the socket to non-blocking mode as a side effect.
   // Our connect() and recv() calls require that the socket be non-blocking.
-  WSAEventSelect(socket_, core_->read_overlapped_.hEvent, FD_CONNECT);
+  if (kUseNonBlockingSockets) {
+    WSAEventSelect(socket_, core_->read_overlapped_.hEvent,
+                   FD_CONNECT | FD_READ | FD_CLOSE);
+  } else {
+    WSAEventSelect(socket_, core_->read_overlapped_.hEvent, FD_CONNECT);
+  }
 
   SockaddrStorage storage;
   if (!endpoint.ToSockAddr(storage.addr, &storage.addr_len))
     return ERR_INVALID_ARGUMENT;
   connect_start_time_ = base::TimeTicks::Now();
   if (!connect(socket_, storage.addr, storage.addr_len)) {
     // Connected without waiting!
     //
     // The MSDN page for connect says:
     //   With a nonblocking socket, the connection attempt cannot be completed
@@ skipping 205 matching lines @@
 
 int TCPClientSocketWin::Read(IOBuffer* buf,
                              int buf_len,
                              const CompletionCallback& callback) {
   DCHECK(CalledOnValidThread());
   DCHECK_NE(socket_, INVALID_SOCKET);
   DCHECK(!waiting_read_);
   DCHECK(read_callback_.is_null());
   DCHECK(!core_->read_iobuffer_);
 
-  buf_len = core_->ThrottleReadSize(buf_len);
+  if (kUseNonBlockingSockets) {
+    core_->read_buffer_.len = buf_len;
+    core_->read_buffer_.buf = buf->data();
 
-  core_->read_buffer_.len = buf_len;
-  core_->read_buffer_.buf = buf->data();
-
-  // TODO(wtc): Remove the assertion after enough testing.
-  AssertEventNotSignaled(core_->read_overlapped_.hEvent);
-  DWORD num, flags = 0;
-  int rv = WSARecv(socket_, &core_->read_buffer_, 1, &num, &flags,
-                   &core_->read_overlapped_, NULL);
-  if (rv == 0) {
-    if (ResetEventIfSignaled(core_->read_overlapped_.hEvent)) {
+    int flags = 0;
+    int rv = recv(socket_, core_->read_buffer_.buf, core_->read_buffer_.len,
+                  flags);
+    if (rv == SOCKET_ERROR) {
+      int os_error = WSAGetLastError();
+      if (os_error != WSAEWOULDBLOCK) {
+        int net_error = MapSystemError(os_error);
+        net_log_.AddEvent(NetLog::TYPE_SOCKET_READ_ERROR,
+                         CreateNetLogSocketErrorCallback(net_error, os_error));

[slamm_google, 2012/08/30 17:47:25]

Indent one more space.

+        return net_error;
+      }
+    } else {
       base::StatsCounter read_bytes("tcp.read_bytes");
-      read_bytes.Add(num);
-      num_bytes_read_ += num;
-      if (num > 0)
+      if (rv > 0) {
         use_history_.set_was_used_to_convey_data();
-      net_log_.AddByteTransferEvent(NetLog::TYPE_SOCKET_BYTES_RECEIVED, num,
+        read_bytes.Add(rv);
+        num_bytes_read_ += rv;
+      }
+      net_log_.AddByteTransferEvent(NetLog::TYPE_SOCKET_BYTES_RECEIVED, rv,
                                     core_->read_buffer_.buf);
-      return static_cast<int>(num);
+      return rv;
     }
   } else {
-    int os_error = WSAGetLastError();
-    if (os_error != WSA_IO_PENDING) {
-      int net_error = MapSystemError(os_error);
-      net_log_.AddEvent(NetLog::TYPE_SOCKET_READ_ERROR,
-                        CreateNetLogSocketErrorCallback(net_error, os_error));
-      return net_error;
+    buf_len = core_->ThrottleReadSize(buf_len);
+
+    core_->read_buffer_.len = buf_len;
+    core_->read_buffer_.buf = buf->data();
+
+    // TODO(wtc): Remove the assertion after enough testing.
+    AssertEventNotSignaled(core_->read_overlapped_.hEvent);
+    DWORD num, flags = 0;
+    int rv = WSARecv(socket_, &core_->read_buffer_, 1, &num, &flags,

[slamm_google, 2012/08/30 17:47:25]

Add a comment to explain the 1 byte recv? 

[pmeenan, 2012/08/30 17:54:31]

Good catch.  That was left over from my other implementation which used a 1-byte
async read instead of the non-blocking wait.  Reverted it to do the full-length
receive.

[pmeenan, 2012/08/31 15:05:23]

Actually, this is 1 recv buffer, not the length and is part of the original code
:-)  The length is set within the structure itself.

+                     &core_->read_overlapped_, NULL);
+    if (rv == 0) {
+      if (ResetEventIfSignaled(core_->read_overlapped_.hEvent)) {
+        base::StatsCounter read_bytes("tcp.read_bytes");
+        read_bytes.Add(num);
+        num_bytes_read_ += num;
+        if (num > 0)
+          use_history_.set_was_used_to_convey_data();
+        net_log_.AddByteTransferEvent(NetLog::TYPE_SOCKET_BYTES_RECEIVED, num,
+                                      core_->read_buffer_.buf);
+        return static_cast<int>(num);
+      }
+    } else {
+      int os_error = WSAGetLastError();
+      if (os_error != WSA_IO_PENDING) {
+        int net_error = MapSystemError(os_error);
+        net_log_.AddEvent(NetLog::TYPE_SOCKET_READ_ERROR,
+                          CreateNetLogSocketErrorCallback(net_error, os_error));
+        return net_error;
+      }
     }
   }
+  waiting_read_ = true;
   core_->WatchForRead();
-  waiting_read_ = true;
   read_callback_ = callback;
   core_->read_iobuffer_ = buf;
   return ERR_IO_PENDING;
 }
 
 int TCPClientSocketWin::Write(IOBuffer* buf,
                               int buf_len,
                               const CompletionCallback& callback) {
   DCHECK(CalledOnValidThread());
   DCHECK_NE(socket_, INVALID_SOCKET);
@@ skipping 135 matching lines @@
   }
 
   connect_os_error_ = os_error;
   rv = DoConnectLoop(result);
   if (rv != ERR_IO_PENDING) {
     LogConnectCompletion(rv);
     DoReadCallback(rv);
   }
 }
 
+void TCPClientSocketWin::HandleReadSignalled() {
+  DCHECK(waiting_read_);
+  int rv;
+  int os_error = 0;
+  WSANETWORKEVENTS events;
+  rv = WSAEnumNetworkEvents(socket_, core_->read_overlapped_.hEvent,
+                            &events);
+  if (rv != SOCKET_ERROR) {
+    if (events.lNetworkEvents & FD_READ) {
+      os_error = events.iErrorCode[FD_READ_BIT];
+      if (os_error == 0) {
+        int flags = 0;
+        rv = recv(socket_, core_->read_buffer_.buf, core_->read_buffer_.len,
+                      flags);

[slamm_google, 2012/08/30 17:47:25]

Fix indentation.

+        if (rv == SOCKET_ERROR) {
+          os_error = WSAGetLastError();
+        }
+      }
+    } else if (events.lNetworkEvents & FD_CLOSE) {
+      os_error = events.iErrorCode[FD_CLOSE_BIT];
+    } else if (events.lNetworkEvents == 0) {
+      core_->WatchForRead();
+      return;
+    }
+  } else {
+    os_error = WSAGetLastError();
+  }
+  if (os_error != 0 && rv <= 0) {
+    rv = MapSystemError(os_error);
+    net_log_.AddEvent(NetLog::TYPE_SOCKET_READ_ERROR,
+                      CreateNetLogSocketErrorCallback(rv, os_error));
+  }
+  waiting_read_ = false;
+  core_->read_iobuffer_ = NULL;
+  DoReadCallback(rv);
+}
+
 void TCPClientSocketWin::DidCompleteRead() {
   DCHECK(waiting_read_);
   DWORD num_bytes, flags;
   BOOL ok = WSAGetOverlappedResult(socket_, &core_->read_overlapped_,
                                    &num_bytes, FALSE, &flags);
   WSAResetEvent(core_->read_overlapped_.hEvent);
   waiting_read_ = false;
   core_->read_iobuffer_ = NULL;
   int rv;
   if (ok) {
@@ skipping 44 matching lines @@
         use_history_.set_was_used_to_convey_data();
       net_log_.AddByteTransferEvent(NetLog::TYPE_SOCKET_BYTES_SENT, num_bytes,
                                     core_->write_buffer_.buf);
     }
   }
   core_->write_iobuffer_ = NULL;
   DoWriteCallback(rv);
 }
 
 }  // namespace net