UDP: Windows implementation using non-blocking IO

net/udp/udp_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/udp/udp_socket_win.h"
 
 #include <mstcpip.h>
 
 #include "base/basictypes.h"
 #include "base/callback.h"
@@ skipping 26 matching lines @@
 
 namespace net {
 
 // This class encapsulates all the state that has to be preserved as long as
 // there is a network IO operation in progress. If the owner UDPSocketWin
 // is destroyed while an operation is in progress, the Core is detached and it
 // lives until the operation completes and the OS doesn't reference any resource
 // declared on this class anymore.
 class UDPSocketWin::Core : public base::RefCounted<Core> {
  public:
-  explicit Core(UDPSocketWin* socket);
+  explicit Core(UDPSocketWin* socket, bool use_overlapped_io);
 
   // Start watching for the end of a read or write operation.

[rvargas (doing something else), 2015/01/22 23:45:13]

nit: the comment is stale (for nb)

[Alpha Left Google, 2015/01/23 02:29:47]

Done.

   void WatchForRead();
   void WatchForWrite();
 
   // The UDPSocketWin is going away.
   void Detach() { socket_ = NULL; }
 
-  // The separate OVERLAPPED variables for asynchronous operation.
-  OVERLAPPED read_overlapped_;
-  OVERLAPPED write_overlapped_;
+  bool use_overlapped_io() const { return use_overlapped_io_; }
 
-  // The buffers used in Read() and Write().
-  scoped_refptr<IOBuffer> read_iobuffer_;
-  scoped_refptr<IOBuffer> write_iobuffer_;
+  OVERLAPPED* read_overlapped() { return &read_overlapped_; }
+  OVERLAPPED* write_overlapped() { return &write_overlapped_; }
+  WSAEVENT read_event() const { return read_event_; }

[rvargas (doing something else), 2015/01/22 23:45:14]

nit: I think we need some empty lines somewhere here. Maybe between unrelated
pairs.

[Alpha Left Google, 2015/01/23 02:29:47]

Done.

+  WSAEVENT write_event() const { return write_event_; }
+  scoped_refptr<IOBuffer> read_iobuffer() const { return read_iobuffer_; }
+  scoped_refptr<IOBuffer> write_iobuffer() const { return write_iobuffer_; }
+  void set_read_iobuffer(scoped_refptr<IOBuffer> buf, int len) {
+    read_iobuffer_ = buf;
+    read_iobuffer_len_ = len;
+  }
+  void set_write_iobuffer(scoped_refptr<IOBuffer> buf, int len) {
+    write_iobuffer_ = buf;
+    write_iobuffer_len_ = len;
+  }
 
-  // The address storage passed to WSARecvFrom().
-  SockaddrStorage recv_addr_storage_;
+  int read_iobuffer_len() const { return read_iobuffer_len_; }
+  int write_iobuffer_len() const { return write_iobuffer_len_; }
+
+  SockaddrStorage* recv_addr_storage() { return &recv_addr_storage_; }
 
  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 12 matching lines @@
     virtual void OnObjectSignaled(HANDLE object);
 
    private:
     Core* const core_;
   };
 
   ~Core();
 
   // The socket that created this object.
   UDPSocketWin* socket_;
+  const bool use_overlapped_io_;
 
   // |reader_| handles the signals from |read_watcher_|.
   ReadDelegate reader_;
   // |writer_| handles the signals from |write_watcher_|.
   WriteDelegate writer_;
 
   // |read_watcher_| watches for events from Read().
   base::win::ObjectWatcher read_watcher_;
   // |write_watcher_| watches for events from Write();
   base::win::ObjectWatcher write_watcher_;
 
+  // The separate OVERLAPPED variables for asynchronous operation.
+  OVERLAPPED read_overlapped_;
+  OVERLAPPED write_overlapped_;
+
+  // Separate events for non-blocking IO operations.
+  WSAEVENT read_event_;

[rvargas (doing something else), 2015/01/22 23:45:14]

Don't do manual handle management. Use a version of ScopedHandle

[Alpha Left Google, 2015/01/23 02:29:47]

I don't see one for WSAEvent. I also think it's strange that we'll use a
ScopedHandle for read_event_ and write_event_ but have to manually manage it for
read_overlapped_.hEvent. I seems to me coherence (i.e. manage manually just like
read_overlapped_.hEvent) is important here.

[rvargas (doing something else), 2015/01/23 20:05:24]

You can use a ScopedHandle directly, or define

typedef GenericScopedHandle<HandleTraits, VerifierTraits> ScopedWSAEvent;

or use a specific WSAEventTraits class.

or use CreateEvent directly to create a manual-reset, non-signaled,
non-inheritable event

There should be no manual handle management anywhere... but rewriting all our
code is complicated. The least we can do is not write new code with manual
handle management.

I agree with the code consistency argument... but the async code is to be
deleted soon so I would be fine living with inconsistent code for a little
while.

Or you could do xx_overlapped_.hEvent = event_.Handle() and remove manual
closing from the async code.

[Alpha Left Google, 2015/01/27 01:52:26]

In this case I'll leave the handle management in Core untouched. But read_event_
and write_event_ will be moved to UDPSocketWin and use ScopedHandle.

+  WSAEVENT write_event_;
+
+  // The buffers used in Read() and Write().
+  scoped_refptr<IOBuffer> read_iobuffer_;
+  scoped_refptr<IOBuffer> write_iobuffer_;
+  int read_iobuffer_len_;
+  int write_iobuffer_len_;
+
+  // The address storage passed to WSARecvFrom().
+  SockaddrStorage recv_addr_storage_;
+
   DISALLOW_COPY_AND_ASSIGN(Core);
 };
 
-UDPSocketWin::Core::Core(UDPSocketWin* socket)
+UDPSocketWin::Core::Core(UDPSocketWin* socket, bool use_overlapped_io)
     : socket_(socket),
+      use_overlapped_io_(use_overlapped_io),
       reader_(this),
-      writer_(this) {
+      writer_(this),
+      read_iobuffer_len_(0),
+      write_iobuffer_len_(0) {
   memset(&read_overlapped_, 0, sizeof(read_overlapped_));
   memset(&write_overlapped_, 0, sizeof(write_overlapped_));
 
-  read_overlapped_.hEvent = WSACreateEvent();
-  write_overlapped_.hEvent = WSACreateEvent();
+  if (use_overlapped_io) {
+    read_overlapped_.hEvent = WSACreateEvent();
+    write_overlapped_.hEvent = WSACreateEvent();
+  } else {
+    read_event_ = WSACreateEvent();
+    write_event_ = WSACreateEvent();
+  }
 }
 
 UDPSocketWin::Core::~Core() {
   // Make sure the message loop is not watching this object anymore.
   read_watcher_.StopWatching();
   write_watcher_.StopWatching();
 
-  WSACloseEvent(read_overlapped_.hEvent);
-  memset(&read_overlapped_, 0xaf, sizeof(read_overlapped_));
-  WSACloseEvent(write_overlapped_.hEvent);
-  memset(&write_overlapped_, 0xaf, sizeof(write_overlapped_));
+  if (use_overlapped_io_) {
+    WSACloseEvent(read_overlapped_.hEvent);
+    memset(&read_overlapped_, 0xaf, sizeof(read_overlapped_));

[rvargas (doing something else), 2015/01/22 23:45:14]

I know this is the old behavior, but we should not do this. It makes things
harder to debug.

[Alpha Left Google, 2015/01/23 02:29:47]

Done.

+    WSACloseEvent(write_overlapped_.hEvent);
+    memset(&write_overlapped_, 0xaf, sizeof(write_overlapped_));
+  } else {
+    WSACloseEvent(read_event_);
+    WSACloseEvent(write_event_);
+  }
 }
 
 void UDPSocketWin::Core::WatchForRead() {
   // We grab an extra reference because there is an IO operation in progress.
   // Balanced in ReadDelegate::OnObjectSignaled().
   AddRef();

[rvargas (doing something else), 2015/01/22 23:45:14]

I don't think this is needed for non-blocking. In fact, the whole core object is
not needed if there are no async operations.

[Alpha Left Google, 2015/01/23 02:29:47]

Core contains a large amount of member variables that are shared between
overlapped mode and non-blocking mode. I seems cleaner to retain this structure
until we decided to get rid of the overlapped mode.

[rvargas (doing something else), 2015/01/23 20:05:24]

That's true, but it makes the post-experiment cleanup more complicated because
it requires a real code refactoring (as opposed to simply deleting unused code
and some renaming).

Regardless of keeping Core at this point, the comment, and grabbing a reference,
is not needed for the non-blocking case.

If we keep Core, there should be a todo referencing the proper bug, to make sure
the cleanup is thorough.

[Alpha Left Google, 2015/01/27 01:52:26]

Okay I'll keep Core untouched but duplicate neededs bits to UDPSocketWin, it
will make removing the overlapped IO easier to do.

-  read_watcher_.StartWatching(read_overlapped_.hEvent, &reader_);
+  if (use_overlapped_io_)
+    read_watcher_.StartWatching(read_overlapped_.hEvent, &reader_);
+  else
+    read_watcher_.StartWatching(read_event_, &reader_);
 }
 
 void UDPSocketWin::Core::WatchForWrite() {
   // We grab an extra reference because there is an IO operation in progress.
   // Balanced in WriteDelegate::OnObjectSignaled().
   AddRef();
-  write_watcher_.StartWatching(write_overlapped_.hEvent, &writer_);
+  if (use_overlapped_io_)
+    write_watcher_.StartWatching(write_overlapped_.hEvent, &writer_);
+  else
+    write_watcher_.StartWatching(write_event_, &writer_);
 }
 
 void UDPSocketWin::Core::ReadDelegate::OnObjectSignaled(HANDLE object) {
   // TODO(vadimt): Remove ScopedTracker below once crbug.com/418183 is fixed.
   tracked_objects::ScopedTracker tracking_profile(
       FROM_HERE_WITH_EXPLICIT_FUNCTION(
           "UDPSocketWin_Core_ReadDelegate_OnObjectSignaled"));
 
-  DCHECK_EQ(object, core_->read_overlapped_.hEvent);
-  if (core_->socket_)
-    core_->socket_->DidCompleteRead();
+  if (core_->use_overlapped_io()) {
+    DCHECK_EQ(object, core_->read_overlapped()->hEvent);
+    if (core_->socket_)
+      core_->socket_->DidCompleteReadOverlapped();
+  } else {
+    DCHECK_EQ(object, core_->read_event());
+    if (core_->socket_)
+      core_->socket_->DidCompleteReadNonBlocking();

[rvargas (doing something else), 2015/01/22 23:45:14]

There is no pending read (no read is completing)

[Alpha Left Google, 2015/01/23 02:29:47]

Changed the naming to OnReadSignaledNonBlocking.

+  }
 
   core_->Release();
 }
 
 void UDPSocketWin::Core::WriteDelegate::OnObjectSignaled(HANDLE object) {
   // TODO(vadimt): Remove ScopedTracker below once crbug.com/418183 is fixed.
   tracked_objects::ScopedTracker tracking_profile(
       FROM_HERE_WITH_EXPLICIT_FUNCTION(
           "UDPSocketWin_Core_WriteDelegate_OnObjectSignaled"));
 
-  DCHECK_EQ(object, core_->write_overlapped_.hEvent);
-  if (core_->socket_)
-    core_->socket_->DidCompleteWrite();
+  if (core_->use_overlapped_io()) {
+    DCHECK_EQ(object, core_->write_overlapped()->hEvent);
+    if (core_->socket_)
+      core_->socket_->DidCompleteWriteOverlapped();
+  } else {
+    DCHECK_EQ(object, core_->write_event());
+    if (core_->socket_)
+      core_->socket_->DidCompleteWriteNonBlocking();
+  }
 
   core_->Release();
 }
 //-----------------------------------------------------------------------------
 
 QwaveAPI::QwaveAPI() : qwave_supported_(false) {
   HMODULE qwave = LoadLibrary(L"qwave.dll");
   if (!qwave)
     return;
   create_handle_func_ =
@@ skipping 74 matching lines @@
                            net::NetLog* net_log,
                            const net::NetLog::Source& source)
     : socket_(INVALID_SOCKET),
       addr_family_(0),
       is_connected_(false),
       socket_options_(SOCKET_OPTION_MULTICAST_LOOP),
       multicast_interface_(0),
       multicast_time_to_live_(1),
       bind_type_(bind_type),
       rand_int_cb_(rand_int_cb),
+      non_blocking_reads_initialized_(false),
+      non_blocking_writes_initialized_(false),
       recv_from_address_(NULL),
       net_log_(BoundNetLog::Make(net_log, NetLog::SOURCE_UDP_SOCKET)),
       qos_handle_(NULL),
-      qos_flow_id_(0) {
+      qos_flow_id_(0),
+      use_overlapped_io_(true) {
   EnsureWinsockInit();
   net_log_.BeginEvent(NetLog::TYPE_SOCKET_ALIVE,
                       source.ToEventParametersCallback());
   if (bind_type == DatagramSocket::RANDOM_BIND)
     DCHECK(!rand_int_cb.is_null());
 }
 
 UDPSocketWin::~UDPSocketWin() {
   Close();
   net_log_.EndEvent(NetLog::TYPE_SOCKET_ALIVE);
 }
 
 int UDPSocketWin::Open(AddressFamily address_family) {
   DCHECK(CalledOnValidThread());
   DCHECK_EQ(socket_, INVALID_SOCKET);
 
   addr_family_ = ConvertAddressFamily(address_family);
   socket_ = CreatePlatformSocket(addr_family_, SOCK_DGRAM, IPPROTO_UDP);
   if (socket_ == INVALID_SOCKET)
     return MapSystemError(WSAGetLastError());
-  core_ = new Core(this);
+  core_ = new Core(this, use_overlapped_io_);
   return OK;
 }
 
 void UDPSocketWin::Close() {
   DCHECK(CalledOnValidThread());
 
   if (socket_ == INVALID_SOCKET)
     return;
 
   if (qos_handle_) {
@@ skipping 71 matching lines @@
                            int buf_len,
                            IPEndPoint* address,
                            const CompletionCallback& callback) {
   DCHECK(CalledOnValidThread());
   DCHECK_NE(INVALID_SOCKET, socket_);
   CHECK(read_callback_.is_null());
   DCHECK(!recv_from_address_);
   DCHECK(!callback.is_null());  // Synchronous operation not supported.
   DCHECK_GT(buf_len, 0);
 
-  int nread = InternalRecvFrom(buf, buf_len, address);
+  int nread = core_->use_overlapped_io()

[rvargas (doing something else), 2015/01/22 23:45:14]

use_overlapped_io_ ?

I mean, why ask the core about this?

[Alpha Left Google, 2015/01/23 02:29:47]

Done.

+                  ? InternalRecvFromOverlapped(buf, buf_len, address)
+                  : InternalRecvFromNonBlocking(buf, buf_len, address);
   if (nread != ERR_IO_PENDING)
     return nread;
 
   read_callback_ = callback;
   recv_from_address_ = address;
   return ERR_IO_PENDING;
 }
 
 int UDPSocketWin::Write(IOBuffer* buf,
                         int buf_len,
@@ skipping 12 matching lines @@
                                 int buf_len,
                                 const IPEndPoint* address,
                                 const CompletionCallback& callback) {
   DCHECK(CalledOnValidThread());
   DCHECK_NE(INVALID_SOCKET, socket_);
   CHECK(write_callback_.is_null());
   DCHECK(!callback.is_null());  // Synchronous operation not supported.
   DCHECK_GT(buf_len, 0);
   DCHECK(!send_to_address_.get());
 
-  int nwrite = InternalSendTo(buf, buf_len, address);
+  int nwrite = core_->use_overlapped_io()
+                   ? InternalSendToOverlapped(buf, buf_len, address)
+                   : InternalSendToNonBlocking(buf, buf_len, address);
   if (nwrite != ERR_IO_PENDING)
     return nwrite;
 
   if (address)
     send_to_address_.reset(new IPEndPoint(*address));
   write_callback_ = callback;
   return ERR_IO_PENDING;
 }
 
 int UDPSocketWin::Connect(const IPEndPoint& address) {
@@ skipping 135 matching lines @@
 void UDPSocketWin::DoWriteCallback(int rv) {
   DCHECK_NE(rv, ERR_IO_PENDING);
   DCHECK(!write_callback_.is_null());
 
   // since Run may result in Write being called, clear write_callback_ up front.
   CompletionCallback c = write_callback_;
   write_callback_.Reset();
   c.Run(rv);
 }
 
-void UDPSocketWin::DidCompleteRead() {
+void UDPSocketWin::DidCompleteReadOverlapped() {
   DWORD num_bytes, flags;
-  BOOL ok = WSAGetOverlappedResult(socket_, &core_->read_overlapped_,
+  BOOL ok = WSAGetOverlappedResult(socket_, core_->read_overlapped(),
                                    &num_bytes, FALSE, &flags);
-  WSAResetEvent(core_->read_overlapped_.hEvent);
+  WSAResetEvent(core_->read_overlapped()->hEvent);
   int result = ok ? num_bytes : MapSystemError(WSAGetLastError());
   // Convert address.
   if (recv_from_address_ && result >= 0) {
     if (!ReceiveAddressToIPEndpoint(recv_from_address_))
       result = ERR_ADDRESS_INVALID;
   }
-  LogRead(result, core_->read_iobuffer_->data());
-  core_->read_iobuffer_ = NULL;
+  LogRead(result, core_->read_iobuffer()->data());
+  core_->set_read_iobuffer(NULL, 0);
   recv_from_address_ = NULL;
   DoReadCallback(result);
 }
 
+void UDPSocketWin::DidCompleteReadNonBlocking() {

[rvargas (doing something else), 2015/01/22 23:45:13]

follow declaration order.

[rvargas (doing something else), 2015/01/22 23:45:14]

DCHECK(buffer)? And we should dcheck(!buffer) on RecvFrom

[Alpha Left Google, 2015/01/23 02:29:47]

Done.

[Alpha Left Google, 2015/01/23 02:29:47]

Done for the first one.

InternalRecvFromNonBlocking is called by OnReadSiangledNonBlocking and
core_->read_iobuffer() is already defined.

[rvargas (doing something else), 2015/01/23 20:05:24]

Right. That's why I suggested adding a dcehck on RecvFrom which is only called
by the user.

[Alpha Left Google, 2015/01/27 01:52:26]

Acknowledged.

+  WSANETWORKEVENTS network_events;
+  int os_error = 0;
+  int rv = WSAEnumNetworkEvents(socket_, core_->read_event(), &network_events);
+  if (rv == SOCKET_ERROR) {
+    os_error = WSAGetLastError();
+    rv = MapSystemError(os_error);
+    DoReadCallback(rv);
+    return;
+  }
+  if (network_events.lNetworkEvents) {
+    DCHECK(network_events.lNetworkEvents & FD_READ);
+    // If network_events.iErrorCode[FD_READ_BIT] is nonzero, still call

[rvargas (doing something else), 2015/01/22 23:45:14]

I don't really follow this comment.

[Alpha Left Google, 2015/01/23 02:29:47]

This is taken from tcp_socket_win but I updated the comment to make it clearer.

[rvargas (doing something else), 2015/01/23 20:05:24]

ok. It still seems somewhat red-herringish as the "natural" thing to do is to
read after receiving the notification, but I guess someone could think about
"optimizing" the code if there was an error at some point (which really doesn't
predict what recv will do)

[Alpha Left Google, 2015/01/27 01:52:26]

It sounds to me you think the comment is bogus. In that case I'll just remove
that. Reading after the notification is the natural thing to do anyway.

+    // InternalRecvFromNonBlocking() because it reports a more accurate error
+    // code.
+    rv = InternalRecvFromNonBlocking(core_->read_iobuffer().get(),
+                                     core_->read_iobuffer_len(),
+                                     recv_from_address_);
+    if (rv == ERR_IO_PENDING)
+      return;
+  } else {
+    core_->WatchForRead();
+    return;
+  }
+  // Convert address if there's data received.
+  if (recv_from_address_ && rv >= 0) {

[rvargas (doing something else), 2015/01/22 23:45:14]

Wasn't this (and logging the read), already done inside
InternalRecvFromNonBlocking?

[Alpha Left Google, 2015/01/23 02:29:47]

Done.

+    if (!ReceiveAddressToIPEndpoint(recv_from_address_))
+      rv = ERR_ADDRESS_INVALID;
+  }
+  LogRead(rv, core_->read_iobuffer()->data());
+  core_->set_read_iobuffer(NULL, 0);
+  recv_from_address_ = NULL;
+  DoReadCallback(rv);
+}
+
 void UDPSocketWin::LogRead(int result, const char* bytes) const {
   if (result < 0) {
     net_log_.AddEventWithNetErrorCode(NetLog::TYPE_UDP_RECEIVE_ERROR, result);
     return;
   }
 
   if (net_log_.IsLogging()) {
     // Get address for logging, if |address| is NULL.
     IPEndPoint address;
     bool is_address_valid = ReceiveAddressToIPEndpoint(&address);
     net_log_.AddEvent(
         NetLog::TYPE_UDP_BYTES_RECEIVED,
         CreateNetLogUDPDataTranferCallback(
             result, bytes,
             is_address_valid ? &address : NULL));
   }
 
   base::StatsCounter read_bytes("udp.read_bytes");
   read_bytes.Add(result);
   NetworkActivityMonitor::GetInstance()->IncrementBytesReceived(result);
 }
 
-void UDPSocketWin::DidCompleteWrite() {
+void UDPSocketWin::DidCompleteWriteOverlapped() {
   DWORD num_bytes, flags;
-  BOOL ok = WSAGetOverlappedResult(socket_, &core_->write_overlapped_,
+  BOOL ok = WSAGetOverlappedResult(socket_, core_->write_overlapped(),
                                    &num_bytes, FALSE, &flags);
-  WSAResetEvent(core_->write_overlapped_.hEvent);
+  WSAResetEvent(core_->write_overlapped()->hEvent);
   int result = ok ? num_bytes : MapSystemError(WSAGetLastError());
-  LogWrite(result, core_->write_iobuffer_->data(), send_to_address_.get());
+  LogWrite(result, core_->write_iobuffer()->data(), send_to_address_.get());
 
   send_to_address_.reset();
-  core_->write_iobuffer_ = NULL;
+  core_->set_write_iobuffer(NULL, 0);
   DoWriteCallback(result);
 }
 
+void UDPSocketWin::DidCompleteWriteNonBlocking() {
+  WSANETWORKEVENTS network_events;
+  int os_error = 0;
+  int rv = WSAEnumNetworkEvents(socket_, core_->write_event(), &network_events);
+  if (rv == SOCKET_ERROR) {
+    os_error = WSAGetLastError();
+    rv = MapSystemError(os_error);
+    DoWriteCallback(rv);
+    return;
+  } else if (network_events.lNetworkEvents) {
+    DCHECK(network_events.lNetworkEvents & FD_WRITE);
+    // If network_events.iErrorCode[FD_WRITE_BIT] is nonzero, still call
+    // InternalSendtoNonBlocking() because it reports a more accurate error
+    // code.
+    rv = InternalSendToNonBlocking(core_->write_iobuffer().get(),
+                                   core_->write_iobuffer_len(),
+                                   send_to_address_.get());
+    if (rv == ERR_IO_PENDING)
+      return;
+  } else {
+    core_->WatchForWrite();
+    return;
+  }
+  LogWrite(rv, core_->write_iobuffer()->data(), send_to_address_.get());
+  core_->set_write_iobuffer(NULL, 0);
+  send_to_address_.reset();
+  DoWriteCallback(rv);
+}
+
 void UDPSocketWin::LogWrite(int result,
                             const char* bytes,
                             const IPEndPoint* address) const {
   if (result < 0) {
     net_log_.AddEventWithNetErrorCode(NetLog::TYPE_UDP_SEND_ERROR, result);
     return;
   }
 
   if (net_log_.IsLogging()) {
     net_log_.AddEvent(
         NetLog::TYPE_UDP_BYTES_SENT,
         CreateNetLogUDPDataTranferCallback(result, bytes, address));
   }
 
   base::StatsCounter write_bytes("udp.write_bytes");
   write_bytes.Add(result);
   NetworkActivityMonitor::GetInstance()->IncrementBytesSent(result);
 }
 
-int UDPSocketWin::InternalRecvFrom(IOBuffer* buf, int buf_len,
-                                   IPEndPoint* address) {
-  DCHECK(!core_->read_iobuffer_.get());
-  SockaddrStorage& storage = core_->recv_addr_storage_;
-  storage.addr_len = sizeof(storage.addr_storage);
+int UDPSocketWin::InternalRecvFromOverlapped(IOBuffer* buf,
+                                             int buf_len,
+                                             IPEndPoint* address) {
+  DCHECK(!core_->read_iobuffer());
+  SockaddrStorage* storage = core_->recv_addr_storage();
+  storage->addr_len = sizeof(storage->addr_storage);
 
   WSABUF read_buffer;
   read_buffer.buf = buf->data();
   read_buffer.len = buf_len;
 
   DWORD flags = 0;
   DWORD num;
   CHECK_NE(INVALID_SOCKET, socket_);
-  AssertEventNotSignaled(core_->read_overlapped_.hEvent);
-  int rv = WSARecvFrom(socket_, &read_buffer, 1, &num, &flags, storage.addr,
-                       &storage.addr_len, &core_->read_overlapped_, NULL);
+  AssertEventNotSignaled(core_->read_overlapped()->hEvent);
+  int rv = WSARecvFrom(socket_, &read_buffer, 1, &num, &flags, storage->addr,
+                       &storage->addr_len, core_->read_overlapped(), NULL);
   if (rv == 0) {
-    if (ResetEventIfSignaled(core_->read_overlapped_.hEvent)) {
+    if (ResetEventIfSignaled(core_->read_overlapped()->hEvent)) {
       int result = num;
       // Convert address.
       if (address && result >= 0) {
         if (!ReceiveAddressToIPEndpoint(address))
           result = ERR_ADDRESS_INVALID;
       }
       LogRead(result, buf->data());
       return result;
     }
   } else {
     int os_error = WSAGetLastError();
     if (os_error != WSA_IO_PENDING) {
       int result = MapSystemError(os_error);
       LogRead(result, NULL);
       return result;
     }
   }
   core_->WatchForRead();
-  core_->read_iobuffer_ = buf;
+  core_->set_read_iobuffer(buf, buf_len);
   return ERR_IO_PENDING;
 }
 
-int UDPSocketWin::InternalSendTo(IOBuffer* buf, int buf_len,
-                                 const IPEndPoint* address) {
-  DCHECK(!core_->write_iobuffer_.get());
+int UDPSocketWin::InternalSendToOverlapped(IOBuffer* buf,
+                                           int buf_len,
+                                           const IPEndPoint* address) {
+  DCHECK(!core_->write_iobuffer());
   SockaddrStorage storage;
   struct sockaddr* addr = storage.addr;
   // Convert address.
   if (!address) {
     addr = NULL;
     storage.addr_len = 0;
   } else {
     if (!address->ToSockAddr(addr, &storage.addr_len)) {
       int result = ERR_ADDRESS_INVALID;
       LogWrite(result, NULL, NULL);
       return result;
     }
   }
 
   WSABUF write_buffer;
   write_buffer.buf = buf->data();
   write_buffer.len = buf_len;
 
   DWORD flags = 0;
   DWORD num;
-  AssertEventNotSignaled(core_->write_overlapped_.hEvent);
-  int rv = WSASendTo(socket_, &write_buffer, 1, &num, flags,
-                     addr, storage.addr_len, &core_->write_overlapped_, NULL);
+  AssertEventNotSignaled(core_->write_overlapped()->hEvent);
+  int rv = WSASendTo(socket_, &write_buffer, 1, &num, flags, addr,
+                     storage.addr_len, core_->write_overlapped(), NULL);
   if (rv == 0) {
-    if (ResetEventIfSignaled(core_->write_overlapped_.hEvent)) {
+    if (ResetEventIfSignaled(core_->write_overlapped()->hEvent)) {
       int result = num;
       LogWrite(result, buf->data(), address);
       return result;
     }
   } else {
     int os_error = WSAGetLastError();
     if (os_error != WSA_IO_PENDING) {
       int result = MapSystemError(os_error);
       LogWrite(result, NULL, NULL);
       return result;
     }
   }
 
   core_->WatchForWrite();
-  core_->write_iobuffer_ = buf;
+  core_->set_write_iobuffer(buf, 0);
   return ERR_IO_PENDING;
 }
 
+int UDPSocketWin::InternalRecvFromNonBlocking(IOBuffer* buf,
+                                              int buf_len,
+                                              IPEndPoint* address) {
+  if (!non_blocking_reads_initialized_) {
+    // After this call the event will be signaled when asynchronous read is
+    // completed.
+    WSAEventSelect(socket_, core_->read_event(), FD_READ);
+    non_blocking_reads_initialized_ = true;
+  }
+  SockaddrStorage* storage = core_->recv_addr_storage();
+  storage->addr_len = sizeof(storage->addr_storage);
+
+  WSABUF read_buffer;

[rvargas (doing something else), 2015/01/22 23:45:14]

Not needed

[Alpha Left Google, 2015/01/23 02:29:47]

Done.

+  read_buffer.buf = buf->data();
+  read_buffer.len = buf_len;
+
+  CHECK_NE(INVALID_SOCKET, socket_);
+  int rv = recvfrom(socket_, buf->data(), buf_len, 0, storage->addr,
+                    &storage->addr_len);
+  if (rv == SOCKET_ERROR) {
+    int os_error = WSAGetLastError();
+    if (os_error != WSAEWOULDBLOCK) {

[rvargas (doing something else), 2015/01/22 23:45:13]

The code will flow better if processing of wouldblock happens here.

[Alpha Left Google, 2015/01/23 02:29:47]

Done.

+      rv = MapSystemError(os_error);
+      LogRead(rv, NULL);
+      return rv;
+    }
+  } else {
+    // Convert address.
+    if (address && rv >= 0 && !ReceiveAddressToIPEndpoint(address)) {
+      rv = ERR_ADDRESS_INVALID;
+    }
+    LogRead(rv, buf->data());
+    return rv;
+  }
+  core_->WatchForRead();
+  core_->set_read_iobuffer(buf, buf_len);
+  return ERR_IO_PENDING;
+}
+
+int UDPSocketWin::InternalSendToNonBlocking(IOBuffer* buf,
+                                            int buf_len,
+                                            const IPEndPoint* address) {
+  if (!non_blocking_writes_initialized_) {
+    // After this call the event will be signaled when asynchronous write is
+    // completed.
+    WSAEventSelect(socket_, core_->write_event(), FD_WRITE);
+    non_blocking_writes_initialized_ = true;
+  }
+  SockaddrStorage storage;
+  struct sockaddr* addr = storage.addr;
+  // Convert address.
+  if (!address) {

[rvargas (doing something else), 2015/01/22 23:45:14]

nit: if (address)

[Alpha Left Google, 2015/01/23 02:29:47]

Done.

+    addr = NULL;
+    storage.addr_len = 0;
+  } else {
+    if (!address->ToSockAddr(addr, &storage.addr_len)) {
+      int result = ERR_ADDRESS_INVALID;
+      LogWrite(result, NULL, NULL);
+      return result;
+    }
+  }
+
+  WSABUF write_buffer;

[rvargas (doing something else), 2015/01/22 23:45:13]

not needed

[Alpha Left Google, 2015/01/23 02:29:47]

Done.

+  write_buffer.buf = buf->data();
+  write_buffer.len = buf_len;
+
+  int rv = sendto(socket_, buf->data(), buf_len, 0, addr, storage.addr_len);
+  if (rv == SOCKET_ERROR) {
+    int os_error = WSAGetLastError();
+    if (os_error != WSAEWOULDBLOCK) {

[rvargas (doing something else), 2015/01/22 23:45:13]

ditto

[Alpha Left Google, 2015/01/23 02:29:46]

Done.

+      rv = MapSystemError(os_error);
+      LogWrite(rv, NULL, NULL);
+      return rv;
+    }
+  } else {
+    LogWrite(rv, buf->data(), address);
+    return rv;
+  }
+  core_->WatchForWrite();
+  core_->set_write_iobuffer(buf, buf_len);
+  return ERR_IO_PENDING;
+}
+
 int UDPSocketWin::SetMulticastOptions() {
   if (!(socket_options_ & SOCKET_OPTION_MULTICAST_LOOP)) {
     DWORD loop = 0;
     int protocol_level =
         addr_family_ == AF_INET ? IPPROTO_IP : IPPROTO_IPV6;
     int option =
         addr_family_ == AF_INET ? IP_MULTICAST_LOOP: IPV6_MULTICAST_LOOP;
     int rv = setsockopt(socket_, protocol_level, option,
                         reinterpret_cast<const char*>(&loop), sizeof(loop));
     if (rv < 0)
@@ skipping 65 matching lines @@
   for (int i = 0; i < kBindRetries; ++i) {
     int rv = DoBind(IPEndPoint(
         address, static_cast<uint16>(rand_int_cb_.Run(kPortStart, kPortEnd))));
     if (rv == OK || rv != ERR_ADDRESS_IN_USE)
       return rv;
   }
   return DoBind(IPEndPoint(address, 0));
 }
 
 bool UDPSocketWin::ReceiveAddressToIPEndpoint(IPEndPoint* address) const {
-  SockaddrStorage& storage = core_->recv_addr_storage_;
-  return address->FromSockAddr(storage.addr, storage.addr_len);
+  SockaddrStorage* storage = core_->recv_addr_storage();
+  return address->FromSockAddr(storage->addr, storage->addr_len);
 }
 
 int UDPSocketWin::JoinGroup(
     const IPAddressNumber& group_address) const {
   DCHECK(CalledOnValidThread());
   if (!is_connected())
     return ERR_SOCKET_NOT_CONNECTED;
 
   switch (group_address.size()) {
     case kIPv4AddressSize: {
@@ skipping 186 matching lines @@
               0,
               NULL);
 
   return OK;
 }
 
 void UDPSocketWin::DetachFromThread() {
   base::NonThreadSafe::DetachFromThread();
 }
 
+void UDPSocketWin::UseNonBlockingIO() {
+  DCHECK(!core_);
+  use_overlapped_io_ = false;
+}
+
 }  // namespace net