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 36 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),

[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)),
       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 (core_->disable_overlapped_reads_) {
+        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);
-
   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)) {
+  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;
 }
 
 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;
@@ skipping 80 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 = 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