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 134 matching lines @@
  public:
   explicit Core(TCPClientSocketWin* socket);
 
   // Start watching for the end of a read or write operation.
   void WatchForRead();
   void WatchForWrite();
 
   // 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();
   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;
-  }
+  // 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>;
 
   class ReadDelegate : public base::win::ObjectWatcher::Delegate {
    public:
     explicit ReadDelegate(Core* core) : core_(core) {}
     virtual ~ReadDelegate() {}
 
     // base::ObjectWatcher::Delegate methods:
@@ skipping 35 matching lines @@
   // returning data to the application.
   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),
+      write_buffer_length_(0),
+      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)),
       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 22 matching lines @@
   AddRef();
   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 if (core_->disable_overlapped_reads_) {
+      core_->socket_->DidSignalRead();
     } else {
       core_->socket_->DidCompleteRead();
     }
   }
 
   core_->Release();
 }
 
 void TCPClientSocketWin::Core::WriteDelegate::OnObjectSignaled(
     HANDLE object) {
@@ skipping 34 matching lines @@
   DCHECK_EQ(socket_, INVALID_SOCKET);
 
   int error = SetupSocket(socket);
   if (error)
     return MapSystemError(error);
 
   socket_ = socket;
   SetNonBlocking(socket_);
 
   core_ = new Core(this);
-
   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.
     return ERR_UNEXPECTED;
@@ skipping 336 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)) {
-      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;
-  return ERR_IO_PENDING;
+  return DoRead(buf, buf_len, callback);
 }
 
 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());
   return SetSocketSendBufferSize(socket_, size);
 }
 
 bool TCPClientSocketWin::SetKeepAlive(bool enable, int delay) {
   return SetTCPKeepAlive(socket_, enable, delay);
 }
 
 bool TCPClientSocketWin::SetNoDelay(bool no_delay) {
   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);
 
   if (net_error != OK) {
     net_log_.EndEventWithNetErrorCode(NetLog::TYPE_TCP_CONNECT, net_error);
     return;
   }
 
   struct sockaddr_storage source_address;
   socklen_t addrlen = sizeof(source_address);
   int rv = getsockname(
       socket_, reinterpret_cast<struct sockaddr*>(&source_address), &addrlen);
   if (rv != 0) {
     LOG(ERROR) << "getsockname() [rv: " << rv
                << "] error: " << WSAGetLastError();
     NOTREACHED();
     net_log_.EndEventWithNetErrorCode(NetLog::TYPE_TCP_CONNECT, rv);
     return;
   }
 
   net_log_.EndEvent(
       NetLog::TYPE_TCP_CONNECT,
       CreateNetLogSourceAddressCallback(
           reinterpret_cast<const struct sockaddr*>(&source_address),
           sizeof(source_address)));
 }
 
+int TCPClientSocketWin::DoRead(IOBuffer* buf, int buf_len,
+                               const CompletionCallback& callback) {
+  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");
+      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,
+                                    buf->data());
+      return rv;
+    }
+  } else {
+    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;
+      }
+    }
+  }
+
+  waiting_read_ = true;
+  read_callback_ = callback;
+  core_->read_iobuffer_ = buf;
+  core_->read_buffer_length_ = buf_len;
+  core_->WatchForRead();
+  return ERR_IO_PENDING;
+}
+
 void TCPClientSocketWin::DoReadCallback(int rv) {
   DCHECK_NE(rv, ERR_IO_PENDING);
   DCHECK(!read_callback_.is_null());
 
   // Since Run may result in Read being called, clear read_callback_ up front.
   CompletionCallback c = read_callback_;
   read_callback_.Reset();
   c.Run(rv);
 }
 
@@ skipping 52 matching lines @@
     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;
+  core_->read_buffer_length_ = 0;
   DoReadCallback(rv);
 }
 
 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);
@@ skipping 19 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::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 = DoRead(core_->read_iobuffer_, core_->read_buffer_length_,
+                read_callback_);
+    if (rv == ERR_IO_PENDING)
+      return;
+  } 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