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

net/socket/tcp_client_socket_win.cc

Index: net/socket/tcp_client_socket_win.cc
===================================================================
--- net/socket/tcp_client_socket_win.cc	(revision 161340)
+++ net/socket/tcp_client_socket_win.cc	(working copy)
@@ -188,9 +188,13 @@
   OVERLAPPED read_overlapped_;
   OVERLAPPED write_overlapped_;
 
+  // Flag indicating if we are in the experiment that enables non-blocking read
+  static bool disable_overlapped_reads_;

[Ryan Sleevi, 2012/10/11 18:53:16]

No need to make this a class static. Just have it be a bool in an unnamed
namespace in the current translation unit (eg: Line 31)

It should follow the Chromium global naming style (g_disable_overlapped_reads)

+
   // The buffers used in Read() and Write().
   scoped_refptr<IOBuffer> read_iobuffer_;
   scoped_refptr<IOBuffer> write_iobuffer_;
+  int read_buffer_length_;
   int write_buffer_length_;
 
   // Throttle the read size based on our current slow start state.
@@ -255,9 +259,12 @@
   DISALLOW_COPY_AND_ASSIGN(Core);
 };
 
+bool TCPClientSocketWin::Core::disable_overlapped_reads_ = false;
+
 TCPClientSocketWin::Core::Core(
     TCPClientSocketWin* socket)
     : write_buffer_length_(0),
+      read_buffer_length_(0),
       socket_(socket),
       ALLOW_THIS_IN_INITIALIZER_LIST(reader_(this)),
       ALLOW_THIS_IN_INITIALIZER_LIST(writer_(this)),
@@ -301,7 +308,11 @@
     if (core_->socket_->waiting_connect()) {
       core_->socket_->DidCompleteConnect();
     } else {
-      core_->socket_->DidCompleteRead();
+      if (disable_overlapped_reads_) {
+        core_->socket_->DidSignalRead();
+      } else {
+        core_->socket_->DidCompleteRead();
+      }
     }
   }
 
@@ -354,6 +365,10 @@
   SetNonBlocking(socket_);
 
   core_ = new Core(this);
+  if (core_->disable_overlapped_reads_) {
+    WSAEventSelect(socket_, core_->read_overlapped_.hEvent,
+                   FD_READ | FD_CLOSE);
+  }
 
   current_address_index_ = 0;
   use_history_.set_was_ever_connected();
@@ -485,7 +500,12 @@
   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 (core_->disable_overlapped_reads_) {
+    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))
@@ -711,41 +731,67 @@
   DCHECK(read_callback_.is_null());
   DCHECK(!core_->read_iobuffer_);
 
-  buf_len = core_->ThrottleReadSize(buf_len);
-
-  WSABUF read_buffer;
-  read_buffer.len = buf_len;
-  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_, &read_buffer, 1, &num, &flags,
-                   &core_->read_overlapped_, NULL);
-  if (rv == 0) {
-    if (ResetEventIfSignaled(core_->read_overlapped_.hEvent)) {
+  if (core_->disable_overlapped_reads_) {
+    int flags = 0;
+    int rv = recv(socket_, buf->data(), buf_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));
+        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,
                                     buf->data());
-      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);
+
+    WSABUF read_buffer;
+    read_buffer.len = buf_len;
+    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_, &read_buffer, 1, &num, &flags,
+                     &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,
+                                      buf->data());
+        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;
+      }
     }
   }
-  core_->WatchForRead();
+
   waiting_read_ = true;
   read_callback_ = callback;
   core_->read_iobuffer_ = buf;
+  core_->read_buffer_length_ = buf_len;
+  core_->WatchForRead();
   return ERR_IO_PENDING;
 }
 
@@ -765,7 +811,6 @@
   WSABUF write_buffer;
   write_buffer.len = buf_len;
   write_buffer.buf = buf->data();
-  core_->write_buffer_length_ = buf_len;
 
   // TODO(wtc): Remove the assertion after enough testing.
   AssertEventNotSignaled(core_->write_overlapped_.hEvent);
@@ -799,10 +844,11 @@
       return net_error;
     }
   }
-  core_->WatchForWrite();
   waiting_write_ = true;
   write_callback_ = callback;
   core_->write_iobuffer_ = buf;
+  core_->write_buffer_length_ = buf_len;
+  core_->WatchForWrite();
   return ERR_IO_PENDING;
 }
 
@@ -900,6 +946,54 @@
   }
 }
 
+void TCPClientSocketWin::DidSignalRead() {
+  DCHECK(waiting_read_);
+  int rv;
+  int os_error = 0, flags = 0;
+  WSANETWORKEVENTS network_events;
+  rv = WSAEnumNetworkEvents(socket_, core_->read_overlapped_.hEvent,
+                              &network_events);
+  if (rv != SOCKET_ERROR) {
+    if (network_events.lNetworkEvents & FD_READ) {
+      rv = recv(socket_, core_->read_iobuffer_->data(),
+                core_->read_buffer_length_, flags);
+      if (rv == SOCKET_ERROR) {
+        os_error = WSAGetLastError();
+        if (os_error == WSAEWOULDBLOCK) {
+          core_->WatchForRead();
+          return;
+        } else {
+          rv = MapSystemError(os_error);
+        }
+      }
+    } else {
+      core_->WatchForRead();
+      return;
+    }
+  } else {
+    os_error = WSAGetLastError();
+    rv = MapSystemError(os_error);
+  }
+  if (rv > 0) {
+      base::StatsCounter read_bytes("tcp.read_bytes");
+      if (rv > 0) {
+        use_history_.set_was_used_to_convey_data();
+        read_bytes.Add(rv);
+        num_bytes_read_ += rv;
+      }
+      net_log_.AddByteTransferEvent(NetLog::TYPE_SOCKET_BYTES_RECEIVED, rv,
+                                    core_->read_iobuffer_->data());
+  } else {
+    rv = MapSystemError(os_error);
+    net_log_.AddEvent(NetLog::TYPE_SOCKET_READ_ERROR,
+                      CreateNetLogSocketErrorCallback(rv, os_error));
+  }
+  waiting_read_ = false;
+  core_->read_iobuffer_ = NULL;
+  core_->read_buffer_length_ = 0;
+  DoReadCallback(rv);
+}
+
 void TCPClientSocketWin::DidCompleteRead() {
   DCHECK(waiting_read_);
   DWORD num_bytes, flags;
@@ -963,4 +1057,8 @@
   DoWriteCallback(rv);
 }
 
+void TCPClientSocketWin::DisableOverlappedReads() {
+  TCPClientSocketWin::Core::disable_overlapped_reads_ = true;
+}
+
 }  // namespace net