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;
   }

[wtc, 2012/10/18 23:17:10]

Could you move the ThrottleReadSize() method up to follow
Detach()? The Style Guide recommends that the methods be
declared as a group, followed by the data members. Thanks.

[pmeenan, 2012/10/19 20:49:41]

Done.

 
+  // Remember the state of overlapped reads for the duration of the socket

[wtc, 2012/10/18 23:17:10]

Nit: overlapped reads => g_disable_overlapped_reads

[pmeenan, 2012/10/19 20:49:41]

Done.

+  // based on what it was when the socket was created.
+  bool disable_overlapped_reads_;
+
  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:
     virtual void OnObjectSignaled(HANDLE object);
@@ skipping 35 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),
+      disable_overlapped_reads_(g_disable_overlapped_reads),
       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 {
-      core_->socket_->DidCompleteRead();
+    } else if(core_->socket_->waiting_read_) {

[wtc, 2012/10/18 23:17:10]

Why is it necessary to test core_->socket_->waiting_read_?

Nit: use the getter method for waiting_read_ because we
use the waiting_connect() getter on line 309.

[pmeenan, 2012/10/19 20:49:41]

Done.

+      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);
+  if (core_->disable_overlapped_reads_) {
+    WSAEventSelect(socket_, core_->read_overlapped_.hEvent,
+                   FD_READ | FD_CLOSE);
+  }

[wtc, 2012/10/18 23:17:10]

I think we should wait until the first Read() call to do this.

[pmeenan, 2012/10/19 20:49:41]

Done.

 
   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 (core_->disable_overlapped_reads_) {
+    WSAEventSelect(socket_, core_->read_overlapped_.hEvent,
+                   FD_CONNECT | FD_READ | FD_CLOSE);

[wtc, 2012/10/18 23:17:10]

I think we should wait until the first Read() call to watch
for FD_READ | FD_CLOSE.

[pmeenan, 2012/10/19 20:49:41]

Done.

+  } 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 (core_->disable_overlapped_reads_) {
+    int flags = 0;

[wtc, 2012/10/18 23:17:10]

Nit: just pass 0 to recv().

[pmeenan, 2012/10/19 20:49:41]

Done.

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

[wtc, 2012/10/18 23:17:10]

Please make lines lines 772-775 look like lines 748-752.

[pmeenan, 2012/10/19 20:49:41]

Done.

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

[wtc, 2012/10/18 23:17:10]

Is it important to call core_->WatchForRead() after
everything has been set up?

[pmeenan, 2012/10/19 20:49:41]

Sorry, wasn't 100% sure about the threading model.  If it is guaranteed that all
actions in here will always be on the same thread then it doesn't matter.  The
protection was just to avoid a potential race condition where the read event
would be signaled and processed before the buffers were initialized.

I can switch it back but logically it seems like you wouldn't want to start the
watching until you finished setting everything up.

   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 15 matching lines @@
       rv = ERR_WINSOCK_UNEXPECTED_WRITTEN_BYTES;
     } else {
       base::StatsCounter write_bytes("tcp.write_bytes");
       write_bytes.Add(num_bytes);
       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;

[wtc, 2012/10/18 23:17:10]

Should we also set core_->write_buffer_length_  to 0 here,
as you do on line 978?

   DoWriteCallback(rv);
 }
 
+void TCPClientSocketWin::DidSignalRead() {
+  DCHECK(waiting_read_);
+  int os_error = 0;
+  int flags = 0;

[wtc, 2012/10/18 23:17:10]

Just pass 0 to recv().

[pmeenan, 2012/10/19 20:49:41]

Done.

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

[wtc, 2012/10/18 23:17:10]

Align this with the first argument on the previous line.

[pmeenan, 2012/10/19 20:49:41]

Done.

+  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(),
+                core_->read_buffer_length_, flags);
+      if (rv == SOCKET_ERROR) {
+        os_error = WSAGetLastError();
+        rv = MapSystemError(os_error);

[wtc, 2012/10/18 23:17:10]

Should we check if os_error is WSAEWOULDBLOCK here?
The MSDN page for WSAEventSelect says we should:
http://msdn.microsoft.com/en-us/library/windows/desktop/ms741576(v=vs.85).aspx

  ... Nevertheless, a robust application must be prepared
  for the possibility that the event object is set and it
  issues a Windows Sockets call that returns WSAEWOULDBLOCK
  immediately. ...

[pmeenan, 2012/10/19 20:49:41]

Might as well for robustness.  The kind of flow they described with recv's being
done at multiple points in the code doesn't happen with Chrome so there should
never be a way to signal FD_READ when there isn't data available to read but if
I've learned anything working on windows is that it never behaves like you think
it should.

Done.

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

[wtc, 2012/10/18 23:17:10]

I guess this is because we receive FD_CLOSE when waiting_read_
is false. TCPClientSocketWin::Core::ReadDelegate::OnObjectSignaled
needs to handle that case. I think it can simply call
WSAEnumNetworkEvents (to reset the event object) and then
either record the error code, or just do nothing. The
next Read() or Write() call will return 0 or fail anyway.

[pmeenan, 2012/10/19 20:49:41]

I don't think so because the code at the time had the check for waiting_read
before even allowing the calls to come in here.

We only watch the read event when we're actually waiting for a read so any close
events that happen outside of a read request will not get detected until the
next socket operation anyway.

+      core_->WatchForRead();
+      return;
+    }
+  }
+  waiting_read_ = false;
+  core_->read_iobuffer_ = NULL;
+  core_->read_buffer_length_ = 0;
+  DoReadCallback(rv);
+}
+
 }  // namespace net