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 243 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),
+      use_non_blocking_io_(true),

[rvargas (doing something else), 2015/02/03 03:11:41]

false

[Alpha Left Google, 2015/02/03 23:24:45]

Done.

+      read_iobuffer_len_(0),
+      write_iobuffer_len_(0),
       recv_from_address_(NULL),
       net_log_(BoundNetLog::Make(net_log, NetLog::SOURCE_UDP_SOCKET)),
       qos_handle_(NULL),
       qos_flow_id_(0) {
   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);
+  if (!use_non_blocking_io_ || SetNonBlocking(socket_)) {

[rvargas (doing something else), 2015/02/03 03:11:41]

Failure to set non blocking must be accounted for the experiment.

[Alpha Left Google, 2015/02/03 23:24:46]

guiwei pointed out that WSAEventSelect already sets the socket to non blocking.
I moved this call.

+    core_ = new Core(this);
+  } else {
+    read_write_event_.Set(WSACreateEvent());
+    WSAEventSelect(socket_, read_write_event_.Get(), FD_READ | FD_WRITE);
+  }
   return OK;
 }
 
 void UDPSocketWin::Close() {
   DCHECK(CalledOnValidThread());
 
   if (socket_ == INVALID_SOCKET)
     return;
 
   if (qos_handle_) {
     QwaveAPI::Get().CloseHandle(qos_handle_);
   }
 
   // Zero out any pending read/write callback state.
   read_callback_.Reset();
   recv_from_address_ = NULL;
   write_callback_.Reset();
 
+  read_write_watcher_.StopWatching();
+  read_write_event_.Close();

[rvargas (doing something else), 2015/02/03 03:11:41]

must be called after closesocket (or it may race to use an invalid handle)

[Alpha Left Google, 2015/02/03 23:24:46]

Done.

+
   base::TimeTicks start_time = base::TimeTicks::Now();
   closesocket(socket_);
   UMA_HISTOGRAM_TIMES("Net.UDPSocketWinClose",
                       base::TimeTicks::Now() - start_time);
   socket_ = INVALID_SOCKET;
   addr_family_ = 0;
   is_connected_ = false;
 
-  core_->Detach();
-  core_ = NULL;
+  if (core_) {
+    core_->Detach();
+    core_ = NULL;
+  }
 }
 
 int UDPSocketWin::GetPeerAddress(IPEndPoint* address) const {
   DCHECK(CalledOnValidThread());
   DCHECK(address);
   if (!is_connected())
     return ERR_SOCKET_NOT_CONNECTED;
 
   // TODO(szym): Simplify. http://crbug.com/126152
   if (!remote_address_.get()) {
@@ skipping 43 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_ ? 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_ ? 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 142 matching lines @@
   c.Run(rv);
 }
 
 void UDPSocketWin::DidCompleteRead() {
   DWORD num_bytes, flags;
   BOOL ok = WSAGetOverlappedResult(socket_, &core_->read_overlapped_,
                                    &num_bytes, FALSE, &flags);
   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_))
+  IPEndPoint address;
+  IPEndPoint* address_to_log = NULL;
+  if (result >= 0) {
+    if (address.FromSockAddr(core_->recv_addr_storage_.addr,
+                             core_->recv_addr_storage_.addr_len)) {
+      if (recv_from_address_) {

[rvargas (doing something else), 2015/02/03 20:55:30]

nit: no {}

[Alpha Left Google, 2015/02/03 23:24:45]

Done.

+        *recv_from_address_ = address;
+      }
+      address_to_log = &address;
+    } else {
       result = ERR_ADDRESS_INVALID;
+    }
   }
-  LogRead(result, core_->read_iobuffer_->data());
+  LogRead(result, core_->read_iobuffer_->data(), address_to_log);
   core_->read_iobuffer_ = NULL;
   recv_from_address_ = NULL;
   DoReadCallback(result);
 }
 
-void UDPSocketWin::LogRead(int result, const char* bytes) const {
+void UDPSocketWin::DidCompleteWrite() {
+  DWORD num_bytes, flags;
+  BOOL ok = WSAGetOverlappedResult(socket_, &core_->write_overlapped_,
+                                   &num_bytes, FALSE, &flags);
+  WSAResetEvent(core_->write_overlapped_.hEvent);
+  int result = ok ? num_bytes : MapSystemError(WSAGetLastError());
+  LogWrite(result, core_->write_iobuffer_->data(), send_to_address_.get());
+
+  send_to_address_.reset();
+  core_->write_iobuffer_ = NULL;
+  DoWriteCallback(result);
+}
+
+void UDPSocketWin::OnObjectSignaled(HANDLE object) {
+  DCHECK(object == read_write_event_.Get());
+  WSANETWORKEVENTS network_events;
+  int os_error = 0;
+  int rv =
+      WSAEnumNetworkEvents(socket_, read_write_event_.Get(), &network_events);
+  if (rv == SOCKET_ERROR) {
+    os_error = WSAGetLastError();
+    rv = MapSystemError(os_error);
+    if (read_iobuffer_) {
+      read_iobuffer_ = NULL;
+      read_iobuffer_len_ = 0;
+      recv_from_address_ = NULL;
+      DoReadCallback(rv);
+    }
+    if (write_iobuffer_) {
+      write_iobuffer_ = NULL;
+      write_iobuffer_len_ = 0;
+      send_to_address_.reset();
+      DoWriteCallback(rv);
+    }
+    return;
+  }
+  if ((network_events.lNetworkEvents & FD_READ) && read_iobuffer_) {
+    OnReadSignaled();
+  }
+  if ((network_events.lNetworkEvents & FD_WRITE) && write_iobuffer_) {
+    OnWriteSignaled();
+  }
+
+  // There's still pending read / write. Watch for further events.
+  if (read_iobuffer_ || write_iobuffer_) {
+    WatchForReadWrite();
+  }
+}
+
+void UDPSocketWin::OnReadSignaled() {
+  DCHECK(read_iobuffer_);

[rvargas (doing something else), 2015/02/03 03:11:41]

Either both methods dcheck the buffer or not. (I have no preference about the
dcheck here, given that the condition is verified at line 646)

However, I must insist in verifying that the buffer is not set when we are being
called by the user, specially now that we ignore subtle mismatches between the
buffer and the event being notified.

[Alpha Left Google, 2015/02/03 23:24:45]

Yeah this DCHECK should be removed since line 646 already does the check.

I'll move the DCHECK to a method that is hit earlier.

+  int rv = InternalRecvFromNonBlocking(read_iobuffer_.get(), read_iobuffer_len_,
+                                       recv_from_address_);
+  if (rv == ERR_IO_PENDING)
+    return;
+  read_iobuffer_ = NULL;
+  read_iobuffer_len_ = 0;
+  recv_from_address_ = NULL;
+  DoReadCallback(rv);
+}
+
+void UDPSocketWin::OnWriteSignaled() {
+  int rv = InternalSendToNonBlocking(write_iobuffer_.get(), write_iobuffer_len_,
+                                     send_to_address_.get());
+  if (rv == ERR_IO_PENDING)
+    return;
+  write_iobuffer_ = NULL;
+  write_iobuffer_len_ = 0;
+  send_to_address_.reset();
+  DoWriteCallback(rv);
+}
+
+void UDPSocketWin::WatchForReadWrite() {
+  if (read_write_watcher_.GetWatchedObject() != NULL) {

[rvargas (doing something else), 2015/02/03 03:11:41]

When would this be the case? (red flag: checking a raw handle vs null)

[Alpha Left Google, 2015/02/03 23:24:46]

For example both write and read got WSA_IO_PENDING there will be two calls to
WatchForReadWrite().

The check for NULL comes from this comment:
https://code.google.com/p/chromium/codesearch#chromium/src/base/win/object_wa...

[rvargas (doing something else), 2015/02/03 23:54:33]

The code from the object watcher is quite old... and technically it may be ok to
indicate if there is no watch in progress by returning a null handle, but these
days checking for a null handle is a code smell (raw handle... there should be
an object somewhere with an explicit method to check validity).

If we are waiting for read and the user writes we would be waiting... but I'd
prefer if this code figures out if it is waiting or not without having to look
at the handle tracked by the watcher. Maybe looking at the write buffer when
reading?

+    DCHECK(read_write_watcher_.GetWatchedObject() == read_write_event_.Get());
+    return;
+  }
+  read_write_watcher_.StartWatching(read_write_event_.Get(), this);
+}
+
+void UDPSocketWin::LogRead(int result,
+                           const char* bytes,
+                           const IPEndPoint* address) 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));
+        CreateNetLogUDPDataTranferCallback(result, bytes, address));
   }
 
   base::StatsCounter read_bytes("udp.read_bytes");
   read_bytes.Add(result);
   NetworkActivityMonitor::GetInstance()->IncrementBytesReceived(result);
 }
 
-void UDPSocketWin::DidCompleteWrite() {
-  DWORD num_bytes, flags;
-  BOOL ok = WSAGetOverlappedResult(socket_, &core_->write_overlapped_,
-                                   &num_bytes, FALSE, &flags);
-  WSAResetEvent(core_->write_overlapped_.hEvent);
-  int result = ok ? num_bytes : MapSystemError(WSAGetLastError());
-  LogWrite(result, core_->write_iobuffer_->data(), send_to_address_.get());
-
-  send_to_address_.reset();
-  core_->write_iobuffer_ = NULL;
-  DoWriteCallback(result);
-}
-
 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) {
+int UDPSocketWin::InternalRecvFromOverlapped(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);
 
   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);
   if (rv == 0) {
     if (ResetEventIfSignaled(core_->read_overlapped_.hEvent)) {
       int result = num;
       // Convert address.
-      if (address && result >= 0) {
-        if (!ReceiveAddressToIPEndpoint(address))
+      IPEndPoint address_storage;
+      IPEndPoint* address_to_log = NULL;
+      if (result >= 0) {
+        if (address_storage.FromSockAddr(core_->recv_addr_storage_.addr,
+                                         core_->recv_addr_storage_.addr_len)) {
+          if (address)
+            *address = address_storage;
+          address_to_log = &address_storage;
+        } else {
           result = ERR_ADDRESS_INVALID;
+        }
       }
-      LogRead(result, buf->data());
+      LogRead(result, buf->data(), address_to_log);
       return result;
     }
   } else {
     int os_error = WSAGetLastError();
     if (os_error != WSA_IO_PENDING) {
       int result = MapSystemError(os_error);
-      LogRead(result, NULL);
+      LogRead(result, NULL, NULL);
       return result;
     }
   }
   core_->WatchForRead();
   core_->read_iobuffer_ = buf;
   return ERR_IO_PENDING;
 }
 
-int UDPSocketWin::InternalSendTo(IOBuffer* buf, int buf_len,
-                                 const IPEndPoint* address) {
+int UDPSocketWin::InternalSendToOverlapped(IOBuffer* buf,
+                                           int buf_len,
+                                           const IPEndPoint* address) {
   DCHECK(!core_->write_iobuffer_.get());
   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;
@@ skipping 24 matching lines @@
       LogWrite(result, NULL, NULL);
       return result;
     }
   }
 
   core_->WatchForWrite();
   core_->write_iobuffer_ = buf;
   return ERR_IO_PENDING;
 }
 
+int UDPSocketWin::InternalRecvFromNonBlocking(IOBuffer* buf,
+                                              int buf_len,
+                                              IPEndPoint* address) {
+  SockaddrStorage storage;
+  storage.addr_len = sizeof(storage.addr_storage);
+
+  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) {
+      read_iobuffer_ = buf;
+      read_iobuffer_len_ = buf_len;
+      WatchForReadWrite();
+      return ERR_IO_PENDING;
+    }
+    rv = MapSystemError(os_error);
+    LogRead(rv, NULL, NULL);
+    return rv;
+  }
+  IPEndPoint address_storage;
+  IPEndPoint* address_to_log = NULL;
+  if (rv >= 0) {
+    if (address_storage.FromSockAddr(storage.addr, storage.addr_len)) {
+      if (address)
+        *address = address_storage;
+      address_to_log = &address_storage;
+    } else {
+      rv = ERR_ADDRESS_INVALID;
+    }
+  }
+  LogRead(rv, buf->data(), address_to_log);
+  return rv;
+}
+
+int UDPSocketWin::InternalSendToNonBlocking(IOBuffer* buf,
+                                            int buf_len,
+                                            const IPEndPoint* address) {
+  SockaddrStorage storage;
+  struct sockaddr* addr = storage.addr;
+  // Convert address.
+  if (address) {
+    if (!address->ToSockAddr(addr, &storage.addr_len)) {
+      int result = ERR_ADDRESS_INVALID;
+      LogWrite(result, NULL, NULL);
+      return result;
+    }
+  } else {
+    addr = NULL;
+    storage.addr_len = 0;
+  }
+
+  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) {
+      write_iobuffer_ = buf;
+      write_iobuffer_len_ = buf_len;
+      WatchForReadWrite();
+      return ERR_IO_PENDING;
+    }
+    rv = MapSystemError(os_error);
+    LogWrite(rv, NULL, NULL);
+    return rv;
+  }
+  LogWrite(rv, buf->data(), address);
+  return rv;
+}
+
 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 64 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);
-}
-
 int UDPSocketWin::JoinGroup(
     const IPAddressNumber& group_address) const {
   DCHECK(CalledOnValidThread());
   if (!is_connected())
     return ERR_SOCKET_NOT_CONNECTED;
 
   switch (group_address.size()) {
     case kIPv4AddressSize: {
       if (addr_family_ != AF_INET)
         return ERR_ADDRESS_INVALID;
@@ skipping 184 matching lines @@
               0,
               NULL);
 
   return OK;
 }
 
 void UDPSocketWin::DetachFromThread() {
   base::NonThreadSafe::DetachFromThread();
 }
 
+void UDPSocketWin::UseNonBlockingIO() {
+  if (core_)
+    return;

[rvargas (doing something else), 2015/02/03 03:11:41]

This looks like a serious issue. The data from the experiment will be wrong.

[Alpha Left Google, 2015/02/03 23:24:46]

I'll return a boolean with this call.

+  use_non_blocking_io_ = true;
+}
+
 }  // namespace net