UDP: Windows implementation using non-blocking IO

net/udp/udp_socket_win.cc

Index: net/udp/udp_socket_win.cc
diff --git a/net/udp/udp_socket_win.cc b/net/udp/udp_socket_win.cc
index 2f5549f8ed2bd16190ddca69c7f4ce746a63b5d6..361177f1066526ff362d16f27ad594a2f1e1e011 100644
--- a/net/udp/udp_socket_win.cc
+++ b/net/udp/udp_socket_win.cc
@@ -44,7 +44,7 @@ namespace net {
 // 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();
@@ -53,16 +53,27 @@ class UDPSocketWin::Core : public base::RefCounted<Core> {
   // 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_; }
+
+  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 buffers used in Read() and Write().
-  scoped_refptr<IOBuffer> read_iobuffer_;
-  scoped_refptr<IOBuffer> write_iobuffer_;
+  int read_iobuffer_len() const { return read_iobuffer_len_; }
+  int write_iobuffer_len() const { return write_iobuffer_len_; }
 
-  // The address storage passed to WSARecvFrom().
-  SockaddrStorage recv_addr_storage_;
+  SockaddrStorage* recv_addr_storage() { return &recv_addr_storage_; }
 
  private:
   friend class base::RefCounted<Core>;
@@ -95,6 +106,7 @@ class UDPSocketWin::Core : public base::RefCounted<Core> {
 
   // The socket that created this object.
   UDPSocketWin* socket_;
+  const bool use_overlapped_io_;
 
   // |reader_| handles the signals from |read_watcher_|.
   ReadDelegate reader_;
@@ -106,18 +118,43 @@ class UDPSocketWin::Core : public base::RefCounted<Core> {
   // |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() {
@@ -125,24 +162,35 @@ UDPSocketWin::Core::~Core() {
   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) {
@@ -151,9 +199,15 @@ void UDPSocketWin::Core::ReadDelegate::OnObjectSignaled(HANDLE object) {
       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();
 }
@@ -164,9 +218,15 @@ void UDPSocketWin::Core::WriteDelegate::OnObjectSignaled(HANDLE object) {
       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();
 }
@@ -261,10 +321,13 @@ UDPSocketWin::UDPSocketWin(DatagramSocket::BindType bind_type,
       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());
@@ -285,7 +348,7 @@ int UDPSocketWin::Open(AddressFamily 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;
 }
 
@@ -377,7 +440,9 @@ int UDPSocketWin::RecvFrom(IOBuffer* buf,
   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;
 
@@ -410,7 +475,9 @@ int UDPSocketWin::SendToOrWrite(IOBuffer* buf,
   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;
 
@@ -566,23 +633,58 @@ void UDPSocketWin::DoWriteCallback(int rv) {
   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);
@@ -605,19 +707,48 @@ void UDPSocketWin::LogRead(int result, const char* bytes) const {
   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 {
@@ -637,11 +768,12 @@ void UDPSocketWin::LogWrite(int 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();
@@ -650,11 +782,11 @@ int UDPSocketWin::InternalRecvFrom(IOBuffer* buf, int 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) {
@@ -673,13 +805,14 @@ int UDPSocketWin::InternalRecvFrom(IOBuffer* buf, int buf_len,
     }
   }
   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.
@@ -700,11 +833,11 @@ int UDPSocketWin::InternalSendTo(IOBuffer* buf, int 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;
@@ -719,7 +852,90 @@ int UDPSocketWin::InternalSendTo(IOBuffer* buf, int buf_len,
   }
 
   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;
 }
 
@@ -808,8 +1024,8 @@ int UDPSocketWin::RandomBind(const IPAddressNumber& address) {
 }
 
 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(
@@ -1016,4 +1232,9 @@ void UDPSocketWin::DetachFromThread() {
   base::NonThreadSafe::DetachFromThread();
 }
 
+void UDPSocketWin::UseNonBlockingIO() {
+  DCHECK(!core_);
+  use_overlapped_io_ = false;
+}
+
 }  // namespace net