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;
+bool g_disable_overlapped_reads = false;
 
 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 143 matching lines @@
 
   // The separate OVERLAPPED variables for asynchronous operation.
   // |read_overlapped_| is used for both Connect() and Read().
   // |write_overlapped_| is only used for Write();
   OVERLAPPED read_overlapped_;
   OVERLAPPED write_overlapped_;
 
   // 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;
@@ skipping 47 matching lines @@
   static const int kInitialSlowStartThrottle = 1 * 1024;
   static const int kMaxSlowStartThrottle = 32 * kInitialSlowStartThrottle;
   int slow_start_throttle_;
 
   DISALLOW_COPY_AND_ASSIGN(Core);
 };
 
 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)),
       slow_start_throttle_(kInitialSlowStartThrottle) {
   memset(&read_overlapped_, 0, sizeof(read_overlapped_));
   memset(&write_overlapped_, 0, sizeof(write_overlapped_));
 
   read_overlapped_.hEvent = WSACreateEvent();
   write_overlapped_.hEvent = WSACreateEvent();
 }
@@ skipping 23 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 (g_disable_overlapped_reads) {

[Ryan Sleevi, 2012/10/12 18:53:13]

BUG? What happens if overlapped reads are disabled after sockets have been
created? It seems like you'll get into a bad state.

My suggestion would be that each TCPClientSocketWin::Core, when created, stores
the value of whatever g_disable_overlapped_reads was, so that for the lifetime
of the socket, its state machine can never become confused.

[pmeenan, 2012/10/17 15:12:31]

Done.

+        core_->socket_->DidSignalRead();
+      } 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 (g_disable_overlapped_reads) {
+    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 (g_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))
     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);
-
-  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 (g_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;
 }
 
 int TCPClientSocketWin::Write(IOBuffer* buf,
                               int buf_len,
                               const CompletionCallback& callback) {
   DCHECK(CalledOnValidThread());
   DCHECK_NE(socket_, INVALID_SOCKET);
   DCHECK(!waiting_write_);
   DCHECK(write_callback_.is_null());
   DCHECK_GT(buf_len, 0);
   DCHECK(!core_->write_iobuffer_);
 
   base::StatsCounter writes("tcp.writes");
   writes.Increment();
 
   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);
   DWORD num;
   int rv = WSASend(socket_, &write_buffer, 1, &num, 0,
                    &core_->write_overlapped_, NULL);
   if (rv == 0) {
     if (ResetEventIfSignaled(core_->write_overlapped_.hEvent)) {
       rv = static_cast<int>(num);
       if (rv > buf_len || rv < 0) {
@@ skipping 13 matching lines @@
     }
   } else {
     int os_error = WSAGetLastError();
     if (os_error != WSA_IO_PENDING) {
       int net_error = MapSystemError(os_error);
       net_log_.AddEvent(NetLog::TYPE_SOCKET_WRITE_ERROR,
                         CreateNetLogSocketErrorCallback(net_error, os_error));
       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;
 }
 
 bool TCPClientSocketWin::SetReceiveBufferSize(int32 size) {
   DCHECK(CalledOnValidThread());
   return SetSocketReceiveBufferSize(socket_, size);
 }
 
 bool TCPClientSocketWin::SetSendBufferSize(int32 size) {
   DCHECK(CalledOnValidThread());
@@ skipping 77 matching lines @@
   }
 
   connect_os_error_ = os_error;
   rv = DoConnectLoop(result);
   if (rv != ERR_IO_PENDING) {
     LogConnectCompletion(rv);
     DoReadCallback(rv);
   }
 }
 
+void TCPClientSocketWin::DidSignalRead() {

[Ryan Sleevi, 2012/10/12 18:53:13]

style nit: The order in the implementation should match the order in the header.

Move this to line 1055 (after DidCompleteWrite)

[pmeenan, 2012/10/17 15:12:31]

Done.

+  DCHECK(waiting_read_);
+  int rv;
+  int os_error = 0, flags = 0;

[Ryan Sleevi, 2012/10/12 18:53:13]

style nit: Chromium style is to have one variable declaration per line.

int os_error = 0;
int flags = 0;

[pmeenan, 2012/10/17 15:12:31]

Done.

+  WSANETWORKEVENTS network_events;
+  rv = WSAEnumNetworkEvents(socket_, core_->read_overlapped_.hEvent,
+                              &network_events);

[Ryan Sleevi, 2012/10/12 18:53:13]

style nit: whitespace

[pmeenan, 2012/10/17 15:12:31]

Done.

+  if (rv != SOCKET_ERROR) {

[rvargas (doing something else), 2012/10/13 03:34:06]

nit: invert the order if the condition.

[pmeenan, 2012/10/17 15:12:31]

Done.

+    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();

[Ryan Sleevi, 2012/10/12 18:53:13]

Do you also need to be watching for FD_CLOSE? Will it be signalled as an FD_READ
of 0 in the event of an unclean close?

If it is signalled as FD_CLOSE, and we ignore it here, does that means we're
potentially holding on indefinitely to a socket that is effectively dead?

ISTM that this conditional should be much more aggressive about handling
expected values and explicitly failing for unhandled values.

[pmeenan, 2012/10/17 15:12:31]

Added the FD_CLOSE handling.  FD_READ does not get signaled in the case of a
close.

+      return;
+    }
+  } else {
+    os_error = WSAGetLastError();
+    rv = MapSystemError(os_error);
+  }
+  if (rv > 0) {
+      base::StatsCounter read_bytes("tcp.read_bytes");

[rvargas (doing something else), 2012/10/13 03:34:06]

This block logically follows the recv call so it would be clear if it's done
before leaving that if()... which points me to the fact that we should extract
the common code that calls recv from here and TCPClientSocketWin::Read().

[pmeenan, 2012/10/17 15:12:31]

This got a lot cleaner when I cleaned it up and it now follows the recv.

+      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);

[rvargas (doing something else), 2012/10/13 03:34:06]

nit: It's a little funky that we can end up here with rv being a system error
and we overwrite it again.

[pmeenan, 2012/10/17 15:12:31]

Should be a lot cleaner now.

+    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;
   BOOL ok = WSAGetOverlappedResult(socket_, &core_->read_overlapped_,
                                    &num_bytes, FALSE, &flags);
   WSAResetEvent(core_->read_overlapped_.hEvent);
   waiting_read_ = false;
   int rv;
   if (ok) {
     base::StatsCounter read_bytes("tcp.read_bytes");
     read_bytes.Add(num_bytes);
     num_bytes_read_ += num_bytes;
     if (num_bytes > 0)
       use_history_.set_was_used_to_convey_data();
     net_log_.AddByteTransferEvent(NetLog::TYPE_SOCKET_BYTES_RECEIVED,
                                   num_bytes, core_->read_iobuffer_->data());
     rv = static_cast<int>(num_bytes);
   } else {
     int os_error = WSAGetLastError();
     rv = MapSystemError(os_error);
     net_log_.AddEvent(NetLog::TYPE_SOCKET_READ_ERROR,
                       CreateNetLogSocketErrorCallback(rv, os_error));
   }
   core_->read_iobuffer_ = NULL;
   DoReadCallback(rv);

[Ryan Sleevi, 2012/10/12 18:53:13]

no core_->read_buffer_length_ = 0?

[pmeenan, 2012/10/17 15:12:31]

Fixed.  The buffer length isn't used in the overlapped I/O code path but it's
cleaner to make sure it is reset and follows the state of the buffer pointer.

 }
 
 void TCPClientSocketWin::DidCompleteWrite() {
   DCHECK(waiting_write_);
 
   DWORD num_bytes, flags;
   BOOL ok = WSAGetOverlappedResult(socket_, &core_->write_overlapped_,
                                    &num_bytes, FALSE, &flags);
   WSAResetEvent(core_->write_overlapped_.hEvent);
   waiting_write_ = false;
@@ skipping 18 matching lines @@
       if (num_bytes > 0)
         use_history_.set_was_used_to_convey_data();
       net_log_.AddByteTransferEvent(NetLog::TYPE_SOCKET_BYTES_SENT, num_bytes,
                                     core_->write_iobuffer_->data());
     }
   }
   core_->write_iobuffer_ = NULL;
   DoWriteCallback(rv);
 }
 
+void TCPClientSocketWin::DisableOverlappedReads() {

[Ryan Sleevi, 2012/10/12 18:53:13]

style nit: The order in the implementation should match the order in the header.

Move this to line 868 (after SetNoDelay)

[pmeenan, 2012/10/17 15:12:31]

Done.

+  g_disable_overlapped_reads = true;
+}
+
 }  // namespace net