Always use NonBlocking IO for UDP sockets on Windows.

net/socket/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/socket/udp_socket_win.h"
 
 #include <mstcpip.h>
 
 #include "base/callback.h"
 #include "base/lazy_instance.h"
@@ skipping 23 matching lines @@
 namespace {
 
 const int kBindRetries = 10;
 const int kPortStart = 1024;
 const int kPortEnd = 65535;
 
 }  // namespace
 
 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);
-
-  // Start watching for the end of a read or write operation.
-  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_;
-
-  // The buffers used in Read() and Write().
-  scoped_refptr<IOBuffer> read_iobuffer_;
-  scoped_refptr<IOBuffer> write_iobuffer_;
-
-  // The address storage passed to WSARecvFrom().
-  SockaddrStorage recv_addr_storage_;
-
- private:
-  friend class base::RefCounted<Core>;
-
-  class ReadDelegate : public base::win::ObjectWatcher::Delegate {
-   public:
-    explicit ReadDelegate(Core* core) : core_(core) {}
-    ~ReadDelegate() override {}
-
-    // base::ObjectWatcher::Delegate methods:
-    void OnObjectSignaled(HANDLE object) override;
-
-   private:
-    Core* const core_;
-  };
-
-  class WriteDelegate : public base::win::ObjectWatcher::Delegate {
-   public:
-    explicit WriteDelegate(Core* core) : core_(core) {}
-    ~WriteDelegate() override {}
-
-    // base::ObjectWatcher::Delegate methods:
-    void OnObjectSignaled(HANDLE object) override;
-
-   private:
-    Core* const core_;
-  };
-
-  ~Core();
-
-  // The socket that created this object.
-  UDPSocketWin* socket_;
-
-  // |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_;
-
-  DISALLOW_COPY_AND_ASSIGN(Core);
-};
-
-UDPSocketWin::Core::Core(UDPSocketWin* socket)
-    : socket_(socket),
-      reader_(this),
-      writer_(this) {
-  memset(&read_overlapped_, 0, sizeof(read_overlapped_));
-  memset(&write_overlapped_, 0, sizeof(write_overlapped_));
-
-  read_overlapped_.hEvent = WSACreateEvent();
-  write_overlapped_.hEvent = 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_));
-}
-
-void UDPSocketWin::Core::WatchForRead() {
-  // We grab an extra reference because there is an IO operation in progress.
-  // Balanced in ReadDelegate::OnObjectSignaled().
-  AddRef();
-  read_watcher_.StartWatchingOnce(read_overlapped_.hEvent, &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_.StartWatchingOnce(write_overlapped_.hEvent, &writer_);
-}
-
-void UDPSocketWin::Core::ReadDelegate::OnObjectSignaled(HANDLE object) {
-  DCHECK_EQ(object, core_->read_overlapped_.hEvent);
-  if (core_->socket_)
-    core_->socket_->DidCompleteRead();
-
-  core_->Release();
-}
-
-void UDPSocketWin::Core::WriteDelegate::OnObjectSignaled(HANDLE object) {
-  DCHECK_EQ(object, core_->write_overlapped_.hEvent);
-  if (core_->socket_)
-    core_->socket_->DidCompleteWrite();
-
-  core_->Release();
-}
-//-----------------------------------------------------------------------------
-
 QwaveAPI::QwaveAPI() : qwave_supported_(false) {
   HMODULE qwave = LoadLibrary(L"qwave.dll");
   if (!qwave)
     return;
   create_handle_func_ =
       (CreateHandleFn)GetProcAddress(qwave, "QOSCreateHandle");
   close_handle_func_ =
       (CloseHandleFn)GetProcAddress(qwave, "QOSCloseHandle");
   add_socket_to_flow_func_ =
       (AddSocketToFlowFn)GetProcAddress(qwave, "QOSAddSocketToFlow");
@@ skipping 69 matching lines @@
                            net::NetLog* net_log,
                            const net::NetLogSource& 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_(false),
       read_iobuffer_len_(0),
       write_iobuffer_len_(0),
       recv_from_address_(NULL),
       net_log_(NetLogWithSource::Make(net_log, NetLogSourceType::UDP_SOCKET)),
       qos_handle_(NULL),
       qos_flow_id_(0) {
   EnsureWinsockInit();
   net_log_.BeginEvent(NetLogEventType::SOCKET_ALIVE,
                       source.ToEventParametersCallback());
   if (bind_type == DatagramSocket::RANDOM_BIND)
     DCHECK(!rand_int_cb.is_null());
 }
 
 UDPSocketWin::~UDPSocketWin() {
   Close();
   net_log_.EndEvent(NetLogEventType::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());
-  if (!use_non_blocking_io_) {
-    core_ = new Core(this);
-  } else {
-    read_write_event_.Set(WSACreateEvent());
-    WSAEventSelect(socket_, read_write_event_.Get(), FD_READ | FD_WRITE);
-  }
+  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();
 
   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;
 
   read_write_watcher_.StopWatching();
   read_write_event_.Close();
-
-  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 45 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 = core_ ? InternalRecvFromOverlapped(buf, buf_len, address)
-                    : InternalRecvFromNonBlocking(buf, buf_len, address);
+  int nread = 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 = core_ ? InternalSendToOverlapped(buf, buf_len, address)
-                     : InternalSendToNonBlocking(buf, buf_len, address);
+  int nwrite = 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 154 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() {
-  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.
-  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_)
-        *recv_from_address_ = address;
-      address_to_log = &address;
-    } else {
-      result = ERR_ADDRESS_INVALID;
-    }
-  }
-  LogRead(result, core_->read_iobuffer_->data(), address_to_log);
-  core_->read_iobuffer_ = NULL;
-  recv_from_address_ = NULL;
-  DoReadCallback(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::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_) {
@@ skipping 81 matching lines @@
 
   if (net_log_.IsCapturing()) {
     net_log_.AddEvent(
         NetLogEventType::UDP_BYTES_SENT,
         CreateNetLogUDPDataTranferCallback(result, bytes, address));
   }
 
   NetworkActivityMonitor::GetInstance()->IncrementBytesSent(result);
 }
 
-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.
-      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(), address_to_log);
-      return result;
-    }
-  } else {
-    int os_error = WSAGetLastError();
-    if (os_error != WSA_IO_PENDING) {
-      int result = MapSystemError(os_error);
-      LogRead(result, NULL, NULL);
-      return result;
-    }
-  }
-  core_->WatchForRead();
-  core_->read_iobuffer_ = buf;
-  return ERR_IO_PENDING;
-}
-
-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;
-      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);
-  if (rv == 0) {
-    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;
-  return ERR_IO_PENDING;
-}
-
 int UDPSocketWin::InternalRecvFromNonBlocking(IOBuffer* buf,
                                               int buf_len,
                                               IPEndPoint* address) {
   DCHECK(!read_iobuffer_ || read_iobuffer_.get() == buf);
   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);
@@ skipping 342 matching lines @@
               0,
               NULL);
 
   return OK;
 }
 
 void UDPSocketWin::DetachFromThread() {
   base::NonThreadSafe::DetachFromThread();
 }
 
-void UDPSocketWin::UseNonBlockingIO() {
-  DCHECK(!core_);
-  use_non_blocking_io_ = true;
-}
-
 }  // namespace net