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 163292)
+++ net/socket/tcp_client_socket_win.cc	(working copy)
@@ -28,6 +28,7 @@
 namespace {
 
 const int kTCPKeepAliveSeconds = 45;
+bool g_disable_overlapped_reads = false;
 
 bool SetSocketReceiveBufferSize(SOCKET socket, int32 size) {
   int rv = setsockopt(socket, SOL_SOCKET, SO_RCVBUF,
@@ -182,6 +183,16 @@
   // The TCPClientSocketWin is going away.
   void Detach() { socket_ = NULL; }
 
+  // Throttle the read size based on our current slow start state.
+  // Returns the throttled read size.
+  int ThrottleReadSize(int size) {
+    if (slow_start_throttle_ < kMaxSlowStartThrottle) {
+      size = std::min(size, slow_start_throttle_);
+      slow_start_throttle_ *= 2;
+    }
+    return size;
+  }
+
   // The separate OVERLAPPED variables for asynchronous operation.
   // |read_overlapped_| is used for both Connect() and Read().
   // |write_overlapped_| is only used for Write();
@@ -191,17 +202,13 @@
   // 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.
-  // Returns the throttled read size.
-  int ThrottleReadSize(int size) {
-    if (slow_start_throttle_ < kMaxSlowStartThrottle) {
-      size = std::min(size, slow_start_throttle_);
-      slow_start_throttle_ *= 2;
-    }
-    return size;
-  }
+  // Remember the state of g_disable_overlapped_reads for the duration of the
+  // socket based on what it was when the socket was created.
+  bool disable_overlapped_reads_;
+  bool non_blocking_reads_initialized_;
 
  private:
   friend class base::RefCounted<Core>;
@@ -258,6 +265,9 @@
 TCPClientSocketWin::Core::Core(
     TCPClientSocketWin* socket)
     : write_buffer_length_(0),
+      read_buffer_length_(0),

[rvargas (doing something else), 2012/10/23 00:19:40]

match the declaration order.

[pmeenan, 2012/10/23 14:29:54]

Done.

+      disable_overlapped_reads_(g_disable_overlapped_reads),
+      non_blocking_reads_initialized_(false),
       socket_(socket),
       ALLOW_THIS_IN_INITIALIZER_LIST(reader_(this)),
       ALLOW_THIS_IN_INITIALIZER_LIST(writer_(this)),
@@ -301,7 +311,11 @@
     if (core_->socket_->waiting_connect()) {
       core_->socket_->DidCompleteConnect();
     } else {
-      core_->socket_->DidCompleteRead();
+      if (core_->disable_overlapped_reads_) {
+        core_->socket_->DidSignalRead();
+      } else {
+        core_->socket_->DidCompleteRead();
+      }
     }
   }
 
@@ -354,7 +368,6 @@
   SetNonBlocking(socket_);
 
   core_ = new Core(this);
-
   current_address_index_ = 0;
   use_history_.set_was_ever_connected();
 
@@ -711,41 +724,73 @@
   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_) {
+    if (!core_->non_blocking_reads_initialized_) {
+      WSAEventSelect(socket_, core_->read_overlapped_.hEvent,
+                     FD_READ | FD_CLOSE);
+      core_->non_blocking_reads_initialized_ = true;
+    }
+    int rv = recv(socket_, buf->data(), buf_len, 0);
+    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;
+    DWORD 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");
+        if (num > 0) {
+          use_history_.set_was_used_to_convey_data();
+          read_bytes.Add(num);
+          num_bytes_read_ += num;
+        }
+        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 +810,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 +843,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;
 }
 
@@ -824,6 +869,10 @@
   return DisableNagle(socket_, no_delay);
 }
 
+void TCPClientSocketWin::DisableOverlappedReads() {
+  g_disable_overlapped_reads = true;
+}
+
 void TCPClientSocketWin::LogConnectCompletion(int net_error) {
   if (net_error == OK)
     UpdateConnectionTypeHistograms(CONNECTION_ANY);
@@ -924,6 +973,7 @@
                       CreateNetLogSocketErrorCallback(rv, os_error));
   }
   core_->read_iobuffer_ = NULL;
+  core_->read_buffer_length_ = 0;
   DoReadCallback(rv);
 }
 
@@ -963,4 +1013,58 @@
   DoWriteCallback(rv);
 }
 
+void TCPClientSocketWin::DidSignalRead() {
+  DCHECK(waiting_read_);
+  int os_error = 0;
+  WSANETWORKEVENTS network_events;
+  int rv = WSAEnumNetworkEvents(socket_, core_->read_overlapped_.hEvent,
+                                &network_events);
+  if (rv == SOCKET_ERROR) {
+    os_error = WSAGetLastError();
+    rv = MapSystemError(os_error);
+  } else {
+    if (network_events.lNetworkEvents & FD_READ) {
+      rv = recv(socket_, core_->read_iobuffer_->data(),

[rvargas (doing something else), 2012/10/23 00:19:40]

The bulk of this condition is basically the same code of Read() (with a little
massaging of the arguments and when to set the waiting_read).

It would be good to either call Read() from here, or extract that common code
into a method that is called from both places.

That said, it is fine to wait until we have results from the field trial (and we
are presumably removing the overlapped code) to do that change.

[pmeenan, 2012/10/23 14:29:54]

Done.

+                core_->read_buffer_length_, 0);
+      if (rv == SOCKET_ERROR) {
+        os_error = WSAGetLastError();
+        if (os_error == WSAEWOULDBLOCK) {
+          core_->WatchForRead();
+          return;
+        } else {
+          rv = MapSystemError(os_error);
+          net_log_.AddEvent(NetLog::TYPE_SOCKET_READ_ERROR,
+                            CreateNetLogSocketErrorCallback(rv, os_error));
+        }
+      } else {
+        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 if (network_events.lNetworkEvents & FD_CLOSE) {
+      if (network_events.iErrorCode[FD_CLOSE_BIT]) {
+        rv = MapSystemError(network_events.iErrorCode[FD_CLOSE_BIT]);
+        net_log_.AddEvent(NetLog::TYPE_SOCKET_READ_ERROR,
+                          CreateNetLogSocketErrorCallback(rv, os_error));
+      } else {
+        rv = 0;
+      }
+    } else {
+      // This should not happen but I have seen cases where we will get
+      // signaled but the network events flags are all clear (0).
+      core_->WatchForRead();
+      return;
+    }
+  }
+  waiting_read_ = false;
+  core_->read_iobuffer_ = NULL;
+  core_->read_buffer_length_ = 0;
+  DoReadCallback(rv);
+}
+
 }  // namespace net