net/udp/udp_socket_win.cc

   1 // Copyright (c) 2012 The Chromium Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style license that can be
   3 // found in the LICENSE file.
   4
   5 #include "net/udp/udp_socket_win.h"
   6
   7 #include <mstcpip.h>
   8
   9 #include "base/basictypes.h"
  10 #include "base/callback.h"
  11 #include "base/lazy_instance.h"
  12 #include "base/logging.h"
  13 #include "base/message_loop/message_loop.h"
  14 #include "base/metrics/histogram.h"
  15 #include "base/metrics/sparse_histogram.h"
  16 #include "base/profiler/scoped_tracker.h"
  17 #include "base/rand_util.h"
  18 #include "net/base/io_buffer.h"
  19 #include "net/base/ip_endpoint.h"
  20 #include "net/base/net_errors.h"
  21 #include "net/base/net_log.h"
  22 #include "net/base/net_util.h"
  23 #include "net/base/network_activity_monitor.h"
  24 #include "net/base/winsock_init.h"
  25 #include "net/base/winsock_util.h"
  26 #include "net/socket/socket_descriptor.h"
  27 #include "net/udp/udp_net_log_parameters.h"
  28
  29 namespace {
  30
  31 const int kBindRetries = 10;
  32 const int kPortStart = 1024;
  33 const int kPortEnd = 65535;
  34
  35 }  // namespace
  36
  37 namespace net {
  38
  39 // This class encapsulates all the state that has to be preserved as long as
  40 // there is a network IO operation in progress. If the owner UDPSocketWin
  41 // is destroyed while an operation is in progress, the Core is detached and it
  42 // lives until the operation completes and the OS doesn't reference any resource
  43 // declared on this class anymore.
  44 class UDPSocketWin::Core : public base::RefCounted<Core> {
  45  public:
  46   explicit Core(UDPSocketWin* socket);
  47
  48   // Start watching for the end of a read or write operation.
  49   void WatchForRead();
  50   void WatchForWrite();
  51
  52   // The UDPSocketWin is going away.
  53   void Detach() { socket_ = NULL; }
  54
  55   // The separate OVERLAPPED variables for asynchronous operation.
  56   OVERLAPPED read_overlapped_;
  57   OVERLAPPED write_overlapped_;
  58
  59   // The buffers used in Read() and Write().
  60   scoped_refptr<IOBuffer> read_iobuffer_;
  61   scoped_refptr<IOBuffer> write_iobuffer_;
  62
  63   // The address storage passed to WSARecvFrom().
  64   SockaddrStorage recv_addr_storage_;
  65
  66  private:
  67   friend class base::RefCounted<Core>;
  68
  69   class ReadDelegate : public base::win::ObjectWatcher::Delegate {
  70    public:
  71     explicit ReadDelegate(Core* core) : core_(core) {}
  72     virtual ~ReadDelegate() {}
  73
  74     // base::ObjectWatcher::Delegate methods:
  75     virtual void OnObjectSignaled(HANDLE object);
  76
  77    private:
  78     Core* const core_;
  79   };
  80
  81   class WriteDelegate : public base::win::ObjectWatcher::Delegate {
  82    public:
  83     explicit WriteDelegate(Core* core) : core_(core) {}
  84     virtual ~WriteDelegate() {}
  85
  86     // base::ObjectWatcher::Delegate methods:
  87     virtual void OnObjectSignaled(HANDLE object);
  88
  89    private:
  90     Core* const core_;
  91   };
  92
  93   ~Core();
  94
  95   // The socket that created this object.
  96   UDPSocketWin* socket_;
  97
  98   // |reader_| handles the signals from |read_watcher_|.
  99   ReadDelegate reader_;
 100   // |writer_| handles the signals from |write_watcher_|.
 101   WriteDelegate writer_;
 102
 103   // |read_watcher_| watches for events from Read().
 104   base::win::ObjectWatcher read_watcher_;
 105   // |write_watcher_| watches for events from Write();
 106   base::win::ObjectWatcher write_watcher_;
 107
 108   DISALLOW_COPY_AND_ASSIGN(Core);
 109 };
 110
 111 UDPSocketWin::Core::Core(UDPSocketWin* socket)
 112     : socket_(socket),
 113       reader_(this),
 114       writer_(this) {
 115   memset(&read_overlapped_, 0, sizeof(read_overlapped_));
 116   memset(&write_overlapped_, 0, sizeof(write_overlapped_));
 117
 118   read_overlapped_.hEvent = WSACreateEvent();
 119   write_overlapped_.hEvent = WSACreateEvent();
 120 }
 121
 122 UDPSocketWin::Core::~Core() {
 123   // Make sure the message loop is not watching this object anymore.
 124   read_watcher_.StopWatching();
 125   write_watcher_.StopWatching();
 126
 127   WSACloseEvent(read_overlapped_.hEvent);
 128   memset(&read_overlapped_, 0xaf, sizeof(read_overlapped_));
 129   WSACloseEvent(write_overlapped_.hEvent);
 130   memset(&write_overlapped_, 0xaf, sizeof(write_overlapped_));
 131 }
 132
 133 void UDPSocketWin::Core::WatchForRead() {
 134   // We grab an extra reference because there is an IO operation in progress.
 135   // Balanced in ReadDelegate::OnObjectSignaled().
 136   AddRef();
 137   read_watcher_.StartWatching(read_overlapped_.hEvent, &reader_);
 138 }
 139
 140 void UDPSocketWin::Core::WatchForWrite() {
 141   // We grab an extra reference because there is an IO operation in progress.
 142   // Balanced in WriteDelegate::OnObjectSignaled().
 143   AddRef();
 144   write_watcher_.StartWatching(write_overlapped_.hEvent, &writer_);
 145 }
 146
 147 void UDPSocketWin::Core::ReadDelegate::OnObjectSignaled(HANDLE object) {
 148   // TODO(pkasting): Remove ScopedTracker below once crbug.com/462789 is fixed.
 149   tracked_objects::ScopedTracker tracking_profile(
 150       FROM_HERE_WITH_EXPLICIT_FUNCTION(
 151           "462789 UDPSocketWin::Core::ReadDelegate::OnObjectSignaled"));
 152
 153   DCHECK_EQ(object, core_->read_overlapped_.hEvent);
 154   if (core_->socket_)
 155     core_->socket_->DidCompleteRead();
 156
 157   core_->Release();
 158 }
 159
 160 void UDPSocketWin::Core::WriteDelegate::OnObjectSignaled(HANDLE object) {
 161   DCHECK_EQ(object, core_->write_overlapped_.hEvent);
 162   if (core_->socket_)
 163     core_->socket_->DidCompleteWrite();
 164
 165   core_->Release();
 166 }
 167 //-----------------------------------------------------------------------------
 168
 169 QwaveAPI::QwaveAPI() : qwave_supported_(false) {
 170   HMODULE qwave = LoadLibrary(L"qwave.dll");
 171   if (!qwave)
 172     return;
 173   create_handle_func_ =
 174       (CreateHandleFn)GetProcAddress(qwave, "QOSCreateHandle");
 175   close_handle_func_ =
 176       (CloseHandleFn)GetProcAddress(qwave, "QOSCloseHandle");
 177   add_socket_to_flow_func_ =
 178       (AddSocketToFlowFn)GetProcAddress(qwave, "QOSAddSocketToFlow");
 179   remove_socket_from_flow_func_ =
 180       (RemoveSocketFromFlowFn)GetProcAddress(qwave, "QOSRemoveSocketFromFlow");
 181   set_flow_func_ = (SetFlowFn)GetProcAddress(qwave, "QOSSetFlow");
 182
 183   if (create_handle_func_ && close_handle_func_ &&
 184       add_socket_to_flow_func_ && remove_socket_from_flow_func_ &&
 185       set_flow_func_) {
 186     qwave_supported_ = true;
 187   }
 188 }
 189
 190 QwaveAPI& QwaveAPI::Get() {
 191   static base::LazyInstance<QwaveAPI>::Leaky lazy_qwave =
 192     LAZY_INSTANCE_INITIALIZER;
 193   return lazy_qwave.Get();
 194 }
 195
 196 bool QwaveAPI::qwave_supported() const {
 197   return qwave_supported_;
 198 }
 199 BOOL QwaveAPI::CreateHandle(PQOS_VERSION version, PHANDLE handle) {
 200   return create_handle_func_(version, handle);
 201 }
 202 BOOL QwaveAPI::CloseHandle(HANDLE handle) {
 203   return close_handle_func_(handle);
 204 }
 205
 206 BOOL QwaveAPI::AddSocketToFlow(HANDLE handle,
 207                                SOCKET socket,
 208                                PSOCKADDR addr,
 209                                QOS_TRAFFIC_TYPE traffic_type,
 210                                DWORD flags,
 211                                PQOS_FLOWID flow_id) {
 212   return add_socket_to_flow_func_(handle,
 213                                   socket,
 214                                   addr,
 215                                   traffic_type,
 216                                   flags,
 217                                   flow_id);
 218 }
 219
 220 BOOL QwaveAPI::RemoveSocketFromFlow(HANDLE handle,
 221                                     SOCKET socket,
 222                                     QOS_FLOWID flow_id,
 223                                     DWORD reserved) {
 224   return remove_socket_from_flow_func_(handle, socket, flow_id, reserved);
 225 }
 226
 227 BOOL QwaveAPI::SetFlow(HANDLE handle,
 228                        QOS_FLOWID flow_id,
 229                        QOS_SET_FLOW op,
 230                        ULONG size,
 231                        PVOID data,
 232                        DWORD reserved,
 233                        LPOVERLAPPED overlapped) {
 234   return set_flow_func_(handle,
 235                         flow_id,
 236                         op,
 237                         size,
 238                         data,
 239                         reserved,
 240                         overlapped);
 241 }
 242
 243
 244 //-----------------------------------------------------------------------------
 245
 246 UDPSocketWin::UDPSocketWin(DatagramSocket::BindType bind_type,
 247                            const RandIntCallback& rand_int_cb,
 248                            net::NetLog* net_log,
 249                            const net::NetLog::Source& source)
 250     : socket_(INVALID_SOCKET),
 251       addr_family_(0),
 252       is_connected_(false),
 253       socket_options_(SOCKET_OPTION_MULTICAST_LOOP),
 254       multicast_interface_(0),
 255       multicast_time_to_live_(1),
 256       bind_type_(bind_type),
 257       rand_int_cb_(rand_int_cb),
 258       use_non_blocking_io_(false),
 259       read_iobuffer_len_(0),
 260       write_iobuffer_len_(0),
 261       recv_from_address_(NULL),
 262       net_log_(BoundNetLog::Make(net_log, NetLog::SOURCE_UDP_SOCKET)),
 263       qos_handle_(NULL),
 264       qos_flow_id_(0) {
 265   EnsureWinsockInit();
 266   net_log_.BeginEvent(NetLog::TYPE_SOCKET_ALIVE,
 267                       source.ToEventParametersCallback());
 268   if (bind_type == DatagramSocket::RANDOM_BIND)
 269     DCHECK(!rand_int_cb.is_null());
 270 }
 271
 272 UDPSocketWin::~UDPSocketWin() {
 273   Close();
 274   net_log_.EndEvent(NetLog::TYPE_SOCKET_ALIVE);
 275 }
 276
 277 int UDPSocketWin::Open(AddressFamily address_family) {
 278   DCHECK(CalledOnValidThread());
 279   DCHECK_EQ(socket_, INVALID_SOCKET);
 280
 281   addr_family_ = ConvertAddressFamily(address_family);
 282   socket_ = CreatePlatformSocket(addr_family_, SOCK_DGRAM, IPPROTO_UDP);
 283   if (socket_ == INVALID_SOCKET)
 284     return MapSystemError(WSAGetLastError());
 285   if (!use_non_blocking_io_) {
 286     core_ = new Core(this);
 287   } else {
 288     read_write_event_.Set(WSACreateEvent());
 289     WSAEventSelect(socket_, read_write_event_.Get(), FD_READ | FD_WRITE);
 290   }
 291   return OK;
 292 }
 293
 294 void UDPSocketWin::Close() {
 295   DCHECK(CalledOnValidThread());
 296
 297   if (socket_ == INVALID_SOCKET)
 298     return;
 299
 300   if (qos_handle_) {
 301     QwaveAPI::Get().CloseHandle(qos_handle_);
 302   }
 303
 304   // Zero out any pending read/write callback state.
 305   read_callback_.Reset();
 306   recv_from_address_ = NULL;
 307   write_callback_.Reset();
 308
 309   base::TimeTicks start_time = base::TimeTicks::Now();
 310   closesocket(socket_);
 311   UMA_HISTOGRAM_TIMES("Net.UDPSocketWinClose",
 312                       base::TimeTicks::Now() - start_time);
 313   socket_ = INVALID_SOCKET;
 314   addr_family_ = 0;
 315   is_connected_ = false;
 316
 317   read_write_watcher_.StopWatching();
 318   read_write_event_.Close();
 319
 320   if (core_) {
 321     core_->Detach();
 322     core_ = NULL;
 323   }
 324 }
 325
 326 int UDPSocketWin::GetPeerAddress(IPEndPoint* address) const {
 327   DCHECK(CalledOnValidThread());
 328   DCHECK(address);
 329   if (!is_connected())
 330     return ERR_SOCKET_NOT_CONNECTED;
 331
 332   // TODO(szym): Simplify. http://crbug.com/126152
 333   if (!remote_address_.get()) {
 334     SockaddrStorage storage;
 335     if (getpeername(socket_, storage.addr, &storage.addr_len))
 336       return MapSystemError(WSAGetLastError());
 337     scoped_ptr<IPEndPoint> remote_address(new IPEndPoint());
 338     if (!remote_address->FromSockAddr(storage.addr, storage.addr_len))
 339       return ERR_ADDRESS_INVALID;
 340     remote_address_.reset(remote_address.release());
 341   }
 342
 343   *address = *remote_address_;
 344   return OK;
 345 }
 346
 347 int UDPSocketWin::GetLocalAddress(IPEndPoint* address) const {
 348   DCHECK(CalledOnValidThread());
 349   DCHECK(address);
 350   if (!is_connected())
 351     return ERR_SOCKET_NOT_CONNECTED;
 352
 353   // TODO(szym): Simplify. http://crbug.com/126152
 354   if (!local_address_.get()) {
 355     SockaddrStorage storage;
 356     if (getsockname(socket_, storage.addr, &storage.addr_len))
 357       return MapSystemError(WSAGetLastError());
 358     scoped_ptr<IPEndPoint> local_address(new IPEndPoint());
 359     if (!local_address->FromSockAddr(storage.addr, storage.addr_len))
 360       return ERR_ADDRESS_INVALID;
 361     local_address_.reset(local_address.release());
 362     net_log_.AddEvent(NetLog::TYPE_UDP_LOCAL_ADDRESS,
 363                       CreateNetLogUDPConnectCallback(local_address_.get()));
 364   }
 365
 366   *address = *local_address_;
 367   return OK;
 368 }
 369
 370 int UDPSocketWin::Read(IOBuffer* buf,
 371                        int buf_len,
 372                        const CompletionCallback& callback) {
 373   return RecvFrom(buf, buf_len, NULL, callback);
 374 }
 375
 376 int UDPSocketWin::RecvFrom(IOBuffer* buf,
 377                            int buf_len,
 378                            IPEndPoint* address,
 379                            const CompletionCallback& callback) {
 380   DCHECK(CalledOnValidThread());
 381   DCHECK_NE(INVALID_SOCKET, socket_);
 382   CHECK(read_callback_.is_null());
 383   DCHECK(!recv_from_address_);
 384   DCHECK(!callback.is_null());  // Synchronous operation not supported.
 385   DCHECK_GT(buf_len, 0);
 386
 387   int nread = core_ ? InternalRecvFromOverlapped(buf, buf_len, address)
 388                     : InternalRecvFromNonBlocking(buf, buf_len, address);
 389   if (nread != ERR_IO_PENDING)
 390     return nread;
 391
 392   read_callback_ = callback;
 393   recv_from_address_ = address;
 394   return ERR_IO_PENDING;
 395 }
 396
 397 int UDPSocketWin::Write(IOBuffer* buf,
 398                         int buf_len,
 399                         const CompletionCallback& callback) {
 400   return SendToOrWrite(buf, buf_len, NULL, callback);
 401 }
 402
 403 int UDPSocketWin::SendTo(IOBuffer* buf,
 404                          int buf_len,
 405                          const IPEndPoint& address,
 406                          const CompletionCallback& callback) {
 407   return SendToOrWrite(buf, buf_len, &address, callback);
 408 }
 409
 410 int UDPSocketWin::SendToOrWrite(IOBuffer* buf,
 411                                 int buf_len,
 412                                 const IPEndPoint* address,
 413                                 const CompletionCallback& callback) {
 414   DCHECK(CalledOnValidThread());
 415   DCHECK_NE(INVALID_SOCKET, socket_);
 416   CHECK(write_callback_.is_null());
 417   DCHECK(!callback.is_null());  // Synchronous operation not supported.
 418   DCHECK_GT(buf_len, 0);
 419   DCHECK(!send_to_address_.get());
 420
 421   int nwrite = core_ ? InternalSendToOverlapped(buf, buf_len, address)
 422                      : InternalSendToNonBlocking(buf, buf_len, address);
 423   if (nwrite != ERR_IO_PENDING)
 424     return nwrite;
 425
 426   if (address)
 427     send_to_address_.reset(new IPEndPoint(*address));
 428   write_callback_ = callback;
 429   return ERR_IO_PENDING;
 430 }
 431
 432 int UDPSocketWin::Connect(const IPEndPoint& address) {
 433   DCHECK_NE(socket_, INVALID_SOCKET);
 434   net_log_.BeginEvent(NetLog::TYPE_UDP_CONNECT,
 435                       CreateNetLogUDPConnectCallback(&address));
 436   int rv = InternalConnect(address);
 437   net_log_.EndEventWithNetErrorCode(NetLog::TYPE_UDP_CONNECT, rv);
 438   is_connected_ = (rv == OK);
 439   return rv;
 440 }
 441
 442 int UDPSocketWin::InternalConnect(const IPEndPoint& address) {
 443   DCHECK(!is_connected());
 444   DCHECK(!remote_address_.get());
 445
 446   int rv = 0;
 447   if (bind_type_ == DatagramSocket::RANDOM_BIND) {
 448     // Construct IPAddressNumber of appropriate size (IPv4 or IPv6) of 0s,
 449     // representing INADDR_ANY or in6addr_any.
 450     size_t addr_size = (address.GetSockAddrFamily() == AF_INET) ?
 451         kIPv4AddressSize : kIPv6AddressSize;
 452     IPAddressNumber addr_any(addr_size);
 453     rv = RandomBind(addr_any);
 454   }
 455   // else connect() does the DatagramSocket::DEFAULT_BIND
 456
 457   if (rv < 0) {
 458     UMA_HISTOGRAM_SPARSE_SLOWLY("Net.UdpSocketRandomBindErrorCode", -rv);
 459     return rv;
 460   }
 461
 462   SockaddrStorage storage;
 463   if (!address.ToSockAddr(storage.addr, &storage.addr_len))
 464     return ERR_ADDRESS_INVALID;
 465
 466   rv = connect(socket_, storage.addr, storage.addr_len);
 467   if (rv < 0)
 468     return MapSystemError(WSAGetLastError());
 469
 470   remote_address_.reset(new IPEndPoint(address));
 471   return rv;
 472 }
 473
 474 int UDPSocketWin::Bind(const IPEndPoint& address) {
 475   DCHECK_NE(socket_, INVALID_SOCKET);
 476   DCHECK(!is_connected());
 477
 478   int rv = SetMulticastOptions();
 479   if (rv < 0)
 480     return rv;
 481
 482   rv = DoBind(address);
 483   if (rv < 0)
 484     return rv;
 485
 486   local_address_.reset();
 487   is_connected_ = true;
 488   return rv;
 489 }
 490
 491 int UDPSocketWin::SetReceiveBufferSize(int32 size) {
 492   DCHECK_NE(socket_, INVALID_SOCKET);
 493   DCHECK(CalledOnValidThread());
 494   int rv = setsockopt(socket_, SOL_SOCKET, SO_RCVBUF,
 495                       reinterpret_cast<const char*>(&size), sizeof(size));
 496   if (rv != 0)
 497     return MapSystemError(WSAGetLastError());
 498
 499   // According to documentation, setsockopt may succeed, but we need to check
 500   // the results via getsockopt to be sure it works on Windows.
 501   int32 actual_size = 0;
 502   int option_size = sizeof(actual_size);
 503   rv = getsockopt(socket_, SOL_SOCKET, SO_RCVBUF,
 504                   reinterpret_cast<char*>(&actual_size), &option_size);
 505   if (rv != 0)
 506     return MapSystemError(WSAGetLastError());
 507   if (actual_size >= size)
 508     return OK;
 509   UMA_HISTOGRAM_CUSTOM_COUNTS("Net.SocketUnchangeableReceiveBuffer",
 510                               actual_size, 1000, 1000000, 50);
 511   return ERR_SOCKET_RECEIVE_BUFFER_SIZE_UNCHANGEABLE;
 512 }
 513
 514 int UDPSocketWin::SetSendBufferSize(int32 size) {
 515   DCHECK_NE(socket_, INVALID_SOCKET);
 516   DCHECK(CalledOnValidThread());
 517   int rv = setsockopt(socket_, SOL_SOCKET, SO_SNDBUF,
 518                       reinterpret_cast<const char*>(&size), sizeof(size));
 519   if (rv != 0)
 520     return MapSystemError(WSAGetLastError());
 521   // According to documentation, setsockopt may succeed, but we need to check
 522   // the results via getsockopt to be sure it works on Windows.
 523   int32 actual_size = 0;
 524   int option_size = sizeof(actual_size);
 525   rv = getsockopt(socket_, SOL_SOCKET, SO_SNDBUF,
 526                   reinterpret_cast<char*>(&actual_size), &option_size);
 527   if (rv != 0)
 528     return MapSystemError(WSAGetLastError());
 529   if (actual_size >= size)
 530     return OK;
 531   UMA_HISTOGRAM_CUSTOM_COUNTS("Net.SocketUnchangeableSendBuffer",
 532                               actual_size, 1000, 1000000, 50);
 533   return ERR_SOCKET_SEND_BUFFER_SIZE_UNCHANGEABLE;
 534 }
 535
 536 int UDPSocketWin::AllowAddressReuse() {
 537   DCHECK_NE(socket_, INVALID_SOCKET);
 538   DCHECK(CalledOnValidThread());
 539   DCHECK(!is_connected());
 540
 541   BOOL true_value = TRUE;
 542   int rv = setsockopt(socket_, SOL_SOCKET, SO_REUSEADDR,
 543                       reinterpret_cast<const char*>(&true_value),
 544                       sizeof(true_value));
 545   return rv == 0 ? OK : MapSystemError(WSAGetLastError());
 546 }
 547
 548 int UDPSocketWin::SetBroadcast(bool broadcast) {
 549   DCHECK_NE(socket_, INVALID_SOCKET);
 550   DCHECK(CalledOnValidThread());
 551
 552   BOOL value = broadcast ? TRUE : FALSE;
 553   int rv = setsockopt(socket_, SOL_SOCKET, SO_BROADCAST,
 554                       reinterpret_cast<const char*>(&value), sizeof(value));
 555   return rv == 0 ? OK : MapSystemError(WSAGetLastError());
 556 }
 557
 558 void UDPSocketWin::DoReadCallback(int rv) {
 559   DCHECK_NE(rv, ERR_IO_PENDING);
 560   DCHECK(!read_callback_.is_null());
 561
 562   // since Run may result in Read being called, clear read_callback_ up front.
 563   CompletionCallback c = read_callback_;
 564   read_callback_.Reset();
 565   c.Run(rv);
 566 }
 567
 568 void UDPSocketWin::DoWriteCallback(int rv) {
 569   DCHECK_NE(rv, ERR_IO_PENDING);
 570   DCHECK(!write_callback_.is_null());
 571
 572   // since Run may result in Write being called, clear write_callback_ up front.
 573   CompletionCallback c = write_callback_;
 574   write_callback_.Reset();
 575   c.Run(rv);
 576 }
 577
 578 void UDPSocketWin::DidCompleteRead() {
 579   DWORD num_bytes, flags;
 580   BOOL ok = WSAGetOverlappedResult(socket_, &core_->read_overlapped_,
 581                                    &num_bytes, FALSE, &flags);
 582   WSAResetEvent(core_->read_overlapped_.hEvent);
 583   int result = ok ? num_bytes : MapSystemError(WSAGetLastError());
 584   // Convert address.
 585   IPEndPoint address;
 586   IPEndPoint* address_to_log = NULL;
 587   if (result >= 0) {
 588     if (address.FromSockAddr(core_->recv_addr_storage_.addr,
 589                              core_->recv_addr_storage_.addr_len)) {
 590       if (recv_from_address_)
 591         *recv_from_address_ = address;
 592       address_to_log = &address;
 593     } else {
 594       result = ERR_ADDRESS_INVALID;
 595     }
 596   }
 597   LogRead(result, core_->read_iobuffer_->data(), address_to_log);
 598   core_->read_iobuffer_ = NULL;
 599   recv_from_address_ = NULL;
 600   DoReadCallback(result);
 601 }
 602
 603 void UDPSocketWin::DidCompleteWrite() {
 604   DWORD num_bytes, flags;
 605   BOOL ok = WSAGetOverlappedResult(socket_, &core_->write_overlapped_,
 606                                    &num_bytes, FALSE, &flags);
 607   WSAResetEvent(core_->write_overlapped_.hEvent);
 608   int result = ok ? num_bytes : MapSystemError(WSAGetLastError());
 609   LogWrite(result, core_->write_iobuffer_->data(), send_to_address_.get());
 610
 611   send_to_address_.reset();
 612   core_->write_iobuffer_ = NULL;
 613   DoWriteCallback(result);
 614 }
 615
 616 void UDPSocketWin::OnObjectSignaled(HANDLE object) {
 617   DCHECK(object == read_write_event_.Get());
 618   WSANETWORKEVENTS network_events;
 619   int os_error = 0;
 620   int rv =
 621       WSAEnumNetworkEvents(socket_, read_write_event_.Get(), &network_events);
 622   if (rv == SOCKET_ERROR) {
 623     os_error = WSAGetLastError();
 624     rv = MapSystemError(os_error);
 625     if (read_iobuffer_) {
 626       read_iobuffer_ = NULL;
 627       read_iobuffer_len_ = 0;
 628       recv_from_address_ = NULL;
 629       DoReadCallback(rv);
 630     }
 631     if (write_iobuffer_) {
 632       write_iobuffer_ = NULL;
 633       write_iobuffer_len_ = 0;
 634       send_to_address_.reset();
 635       DoWriteCallback(rv);
 636     }
 637     return;
 638   }
 639   if ((network_events.lNetworkEvents & FD_READ) && read_iobuffer_) {
 640     OnReadSignaled();
 641   }
 642   if ((network_events.lNetworkEvents & FD_WRITE) && write_iobuffer_) {
 643     OnWriteSignaled();
 644   }
 645
 646   // There's still pending read / write. Watch for further events.
 647   if (read_iobuffer_ || write_iobuffer_) {
 648     WatchForReadWrite();
 649   }
 650 }
 651
 652 void UDPSocketWin::OnReadSignaled() {
 653   int rv = InternalRecvFromNonBlocking(read_iobuffer_.get(), read_iobuffer_len_,
 654                                        recv_from_address_);
 655   if (rv == ERR_IO_PENDING)
 656     return;
 657   read_iobuffer_ = NULL;
 658   read_iobuffer_len_ = 0;
 659   recv_from_address_ = NULL;
 660   DoReadCallback(rv);
 661 }
 662
 663 void UDPSocketWin::OnWriteSignaled() {
 664   int rv = InternalSendToNonBlocking(write_iobuffer_.get(), write_iobuffer_len_,
 665                                      send_to_address_.get());
 666   if (rv == ERR_IO_PENDING)
 667     return;
 668   write_iobuffer_ = NULL;
 669   write_iobuffer_len_ = 0;
 670   send_to_address_.reset();
 671   DoWriteCallback(rv);
 672 }
 673
 674 void UDPSocketWin::WatchForReadWrite() {
 675   if (read_write_watcher_.IsWatching())
 676     return;
 677   bool watched =
 678       read_write_watcher_.StartWatching(read_write_event_.Get(), this);
 679   DCHECK(watched);
 680 }
 681
 682 void UDPSocketWin::LogRead(int result,
 683                            const char* bytes,
 684                            const IPEndPoint* address) const {
 685   if (result < 0) {
 686     net_log_.AddEventWithNetErrorCode(NetLog::TYPE_UDP_RECEIVE_ERROR, result);
 687     return;
 688   }
 689
 690   if (net_log_.IsLogging()) {
 691     net_log_.AddEvent(
 692         NetLog::TYPE_UDP_BYTES_RECEIVED,
 693         CreateNetLogUDPDataTranferCallback(result, bytes, address));
 694   }
 695
 696   NetworkActivityMonitor::GetInstance()->IncrementBytesReceived(result);
 697 }
 698
 699 void UDPSocketWin::LogWrite(int result,
 700                             const char* bytes,
 701                             const IPEndPoint* address) const {
 702   if (result < 0) {
 703     net_log_.AddEventWithNetErrorCode(NetLog::TYPE_UDP_SEND_ERROR, result);
 704     return;
 705   }
 706
 707   if (net_log_.IsLogging()) {
 708     net_log_.AddEvent(
 709         NetLog::TYPE_UDP_BYTES_SENT,
 710         CreateNetLogUDPDataTranferCallback(result, bytes, address));
 711   }
 712
 713   NetworkActivityMonitor::GetInstance()->IncrementBytesSent(result);
 714 }
 715
 716 int UDPSocketWin::InternalRecvFromOverlapped(IOBuffer* buf,
 717                                              int buf_len,
 718                                              IPEndPoint* address) {
 719   DCHECK(!core_->read_iobuffer_.get());
 720   SockaddrStorage& storage = core_->recv_addr_storage_;
 721   storage.addr_len = sizeof(storage.addr_storage);
 722
 723   WSABUF read_buffer;
 724   read_buffer.buf = buf->data();
 725   read_buffer.len = buf_len;
 726
 727   DWORD flags = 0;
 728   DWORD num;
 729   CHECK_NE(INVALID_SOCKET, socket_);
 730   AssertEventNotSignaled(core_->read_overlapped_.hEvent);
 731   int rv = WSARecvFrom(socket_, &read_buffer, 1, &num, &flags, storage.addr,
 732                        &storage.addr_len, &core_->read_overlapped_, NULL);
 733   if (rv == 0) {
 734     if (ResetEventIfSignaled(core_->read_overlapped_.hEvent)) {
 735       int result = num;
 736       // Convert address.
 737       IPEndPoint address_storage;
 738       IPEndPoint* address_to_log = NULL;
 739       if (result >= 0) {
 740         if (address_storage.FromSockAddr(core_->recv_addr_storage_.addr,
 741                                          core_->recv_addr_storage_.addr_len)) {
 742           if (address)
 743             *address = address_storage;
 744           address_to_log = &address_storage;
 745         } else {
 746           result = ERR_ADDRESS_INVALID;
 747         }
 748       }
 749       LogRead(result, buf->data(), address_to_log);
 750       return result;
 751     }
 752   } else {
 753     int os_error = WSAGetLastError();
 754     if (os_error != WSA_IO_PENDING) {
 755       int result = MapSystemError(os_error);
 756       LogRead(result, NULL, NULL);
 757       return result;
 758     }
 759   }
 760   core_->WatchForRead();
 761   core_->read_iobuffer_ = buf;
 762   return ERR_IO_PENDING;
 763 }
 764
 765 int UDPSocketWin::InternalSendToOverlapped(IOBuffer* buf,
 766                                            int buf_len,
 767                                            const IPEndPoint* address) {
 768   DCHECK(!core_->write_iobuffer_.get());
 769   SockaddrStorage storage;
 770   struct sockaddr* addr = storage.addr;
 771   // Convert address.
 772   if (!address) {
 773     addr = NULL;
 774     storage.addr_len = 0;
 775   } else {
 776     if (!address->ToSockAddr(addr, &storage.addr_len)) {
 777       int result = ERR_ADDRESS_INVALID;
 778       LogWrite(result, NULL, NULL);
 779       return result;
 780     }
 781   }
 782
 783   WSABUF write_buffer;
 784   write_buffer.buf = buf->data();
 785   write_buffer.len = buf_len;
 786
 787   DWORD flags = 0;
 788   DWORD num;
 789   AssertEventNotSignaled(core_->write_overlapped_.hEvent);
 790   int rv = WSASendTo(socket_, &write_buffer, 1, &num, flags,
 791                      addr, storage.addr_len, &core_->write_overlapped_, NULL);
 792   if (rv == 0) {
 793     if (ResetEventIfSignaled(core_->write_overlapped_.hEvent)) {
 794       int result = num;
 795       LogWrite(result, buf->data(), address);
 796       return result;
 797     }
 798   } else {
 799     int os_error = WSAGetLastError();
 800     if (os_error != WSA_IO_PENDING) {
 801       int result = MapSystemError(os_error);
 802       LogWrite(result, NULL, NULL);
 803       return result;
 804     }
 805   }
 806
 807   core_->WatchForWrite();
 808   core_->write_iobuffer_ = buf;
 809   return ERR_IO_PENDING;
 810 }
 811
 812 int UDPSocketWin::InternalRecvFromNonBlocking(IOBuffer* buf,
 813                                               int buf_len,
 814                                               IPEndPoint* address) {
 815   DCHECK(!read_iobuffer_ || read_iobuffer_.get() == buf);
 816   SockaddrStorage storage;
 817   storage.addr_len = sizeof(storage.addr_storage);
 818
 819   CHECK_NE(INVALID_SOCKET, socket_);
 820   int rv = recvfrom(socket_, buf->data(), buf_len, 0, storage.addr,
 821                     &storage.addr_len);
 822   if (rv == SOCKET_ERROR) {
 823     int os_error = WSAGetLastError();
 824     if (os_error == WSAEWOULDBLOCK) {
 825       read_iobuffer_ = buf;
 826       read_iobuffer_len_ = buf_len;
 827       WatchForReadWrite();
 828       return ERR_IO_PENDING;
 829     }
 830     rv = MapSystemError(os_error);
 831     LogRead(rv, NULL, NULL);
 832     return rv;
 833   }
 834   IPEndPoint address_storage;
 835   IPEndPoint* address_to_log = NULL;
 836   if (rv >= 0) {
 837     if (address_storage.FromSockAddr(storage.addr, storage.addr_len)) {
 838       if (address)
 839         *address = address_storage;
 840       address_to_log = &address_storage;
 841     } else {
 842       rv = ERR_ADDRESS_INVALID;
 843     }
 844   }
 845   LogRead(rv, buf->data(), address_to_log);
 846   return rv;
 847 }
 848
 849 int UDPSocketWin::InternalSendToNonBlocking(IOBuffer* buf,
 850                                             int buf_len,
 851                                             const IPEndPoint* address) {
 852   DCHECK(!write_iobuffer_ || write_iobuffer_.get() == buf);
 853   SockaddrStorage storage;
 854   struct sockaddr* addr = storage.addr;
 855   // Convert address.
 856   if (address) {
 857     if (!address->ToSockAddr(addr, &storage.addr_len)) {
 858       int result = ERR_ADDRESS_INVALID;
 859       LogWrite(result, NULL, NULL);
 860       return result;
 861     }
 862   } else {
 863     addr = NULL;
 864     storage.addr_len = 0;
 865   }
 866
 867   int rv = sendto(socket_, buf->data(), buf_len, 0, addr, storage.addr_len);
 868   if (rv == SOCKET_ERROR) {
 869     int os_error = WSAGetLastError();
 870     if (os_error == WSAEWOULDBLOCK) {
 871       write_iobuffer_ = buf;
 872       write_iobuffer_len_ = buf_len;
 873       WatchForReadWrite();
 874       return ERR_IO_PENDING;
 875     }
 876     rv = MapSystemError(os_error);
 877     LogWrite(rv, NULL, NULL);
 878     return rv;
 879   }
 880   LogWrite(rv, buf->data(), address);
 881   return rv;
 882 }
 883
 884 int UDPSocketWin::SetMulticastOptions() {
 885   if (!(socket_options_ & SOCKET_OPTION_MULTICAST_LOOP)) {
 886     DWORD loop = 0;
 887     int protocol_level =
 888         addr_family_ == AF_INET ? IPPROTO_IP : IPPROTO_IPV6;
 889     int option =
 890         addr_family_ == AF_INET ? IP_MULTICAST_LOOP: IPV6_MULTICAST_LOOP;
 891     int rv = setsockopt(socket_, protocol_level, option,
 892                         reinterpret_cast<const char*>(&loop), sizeof(loop));
 893     if (rv < 0)
 894       return MapSystemError(WSAGetLastError());
 895   }
 896   if (multicast_time_to_live_ != 1) {
 897     DWORD hops = multicast_time_to_live_;
 898     int protocol_level =
 899         addr_family_ == AF_INET ? IPPROTO_IP : IPPROTO_IPV6;
 900     int option =
 901         addr_family_ == AF_INET ? IP_MULTICAST_TTL: IPV6_MULTICAST_HOPS;
 902     int rv = setsockopt(socket_, protocol_level, option,
 903                         reinterpret_cast<const char*>(&hops), sizeof(hops));
 904     if (rv < 0)
 905       return MapSystemError(WSAGetLastError());
 906   }
 907   if (multicast_interface_ != 0) {
 908     switch (addr_family_) {
 909       case AF_INET: {
 910         in_addr address;
 911         address.s_addr = htonl(multicast_interface_);
 912         int rv = setsockopt(socket_, IPPROTO_IP, IP_MULTICAST_IF,
 913                             reinterpret_cast<const char*>(&address),
 914                             sizeof(address));
 915         if (rv)
 916           return MapSystemError(WSAGetLastError());
 917         break;
 918       }
 919       case AF_INET6: {
 920         uint32 interface_index = multicast_interface_;
 921         int rv = setsockopt(socket_, IPPROTO_IPV6, IPV6_MULTICAST_IF,
 922                             reinterpret_cast<const char*>(&interface_index),
 923                             sizeof(interface_index));
 924         if (rv)
 925           return MapSystemError(WSAGetLastError());
 926         break;
 927       }
 928       default:
 929         NOTREACHED() << "Invalid address family";
 930         return ERR_ADDRESS_INVALID;
 931     }
 932   }
 933   return OK;
 934 }
 935
 936 int UDPSocketWin::DoBind(const IPEndPoint& address) {
 937   SockaddrStorage storage;
 938   if (!address.ToSockAddr(storage.addr, &storage.addr_len))
 939     return ERR_ADDRESS_INVALID;
 940   int rv = bind(socket_, storage.addr, storage.addr_len);
 941   if (rv == 0)
 942     return OK;
 943   int last_error = WSAGetLastError();
 944   UMA_HISTOGRAM_SPARSE_SLOWLY("Net.UdpSocketBindErrorFromWinOS", last_error);
 945   // Map some codes that are special to bind() separately.
 946   // * WSAEACCES: If a port is already bound to a socket, WSAEACCES may be
 947   //   returned instead of WSAEADDRINUSE, depending on whether the socket
 948   //   option SO_REUSEADDR or SO_EXCLUSIVEADDRUSE is set and whether the
 949   //   conflicting socket is owned by a different user account. See the MSDN
 950   //   page "Using SO_REUSEADDR and SO_EXCLUSIVEADDRUSE" for the gory details.
 951   if (last_error == WSAEACCES || last_error == WSAEADDRNOTAVAIL)
 952     return ERR_ADDRESS_IN_USE;
 953   return MapSystemError(last_error);
 954 }
 955
 956 int UDPSocketWin::RandomBind(const IPAddressNumber& address) {
 957   DCHECK(bind_type_ == DatagramSocket::RANDOM_BIND && !rand_int_cb_.is_null());
 958
 959   for (int i = 0; i < kBindRetries; ++i) {
 960     int rv = DoBind(IPEndPoint(
 961         address, static_cast<uint16>(rand_int_cb_.Run(kPortStart, kPortEnd))));
 962     if (rv == OK || rv != ERR_ADDRESS_IN_USE)
 963       return rv;
 964   }
 965   return DoBind(IPEndPoint(address, 0));
 966 }
 967
 968 int UDPSocketWin::JoinGroup(
 969     const IPAddressNumber& group_address) const {
 970   DCHECK(CalledOnValidThread());
 971   if (!is_connected())
 972     return ERR_SOCKET_NOT_CONNECTED;
 973
 974   switch (group_address.size()) {
 975     case kIPv4AddressSize: {
 976       if (addr_family_ != AF_INET)
 977         return ERR_ADDRESS_INVALID;
 978       ip_mreq mreq;
 979       mreq.imr_interface.s_addr = htonl(multicast_interface_);
 980       memcpy(&mreq.imr_multiaddr, &group_address[0], kIPv4AddressSize);
 981       int rv = setsockopt(socket_, IPPROTO_IP, IP_ADD_MEMBERSHIP,
 982                           reinterpret_cast<const char*>(&mreq),
 983                           sizeof(mreq));
 984       if (rv)
 985         return MapSystemError(WSAGetLastError());
 986       return OK;
 987     }
 988     case kIPv6AddressSize: {
 989       if (addr_family_ != AF_INET6)
 990         return ERR_ADDRESS_INVALID;
 991       ipv6_mreq mreq;
 992       mreq.ipv6mr_interface = multicast_interface_;
 993       memcpy(&mreq.ipv6mr_multiaddr, &group_address[0], kIPv6AddressSize);
 994       int rv = setsockopt(socket_, IPPROTO_IPV6, IPV6_ADD_MEMBERSHIP,
 995                           reinterpret_cast<const char*>(&mreq),
 996                           sizeof(mreq));
 997       if (rv)
 998         return MapSystemError(WSAGetLastError());
 999       return OK;
1000     }
1001     default:
1002       NOTREACHED() << "Invalid address family";
1003       return ERR_ADDRESS_INVALID;
1004   }
1005 }
1006
1007 int UDPSocketWin::LeaveGroup(
1008     const IPAddressNumber& group_address) const {
1009   DCHECK(CalledOnValidThread());
1010   if (!is_connected())
1011     return ERR_SOCKET_NOT_CONNECTED;
1012
1013   switch (group_address.size()) {
1014     case kIPv4AddressSize: {
1015       if (addr_family_ != AF_INET)
1016         return ERR_ADDRESS_INVALID;
1017       ip_mreq mreq;
1018       mreq.imr_interface.s_addr = htonl(multicast_interface_);
1019       memcpy(&mreq.imr_multiaddr, &group_address[0], kIPv4AddressSize);
1020       int rv = setsockopt(socket_, IPPROTO_IP, IP_DROP_MEMBERSHIP,
1021                           reinterpret_cast<const char*>(&mreq), sizeof(mreq));
1022       if (rv)
1023         return MapSystemError(WSAGetLastError());
1024       return OK;
1025     }
1026     case kIPv6AddressSize: {
1027       if (addr_family_ != AF_INET6)
1028         return ERR_ADDRESS_INVALID;
1029       ipv6_mreq mreq;
1030       mreq.ipv6mr_interface = multicast_interface_;
1031       memcpy(&mreq.ipv6mr_multiaddr, &group_address[0], kIPv6AddressSize);
1032       int rv = setsockopt(socket_, IPPROTO_IPV6, IP_DROP_MEMBERSHIP,
1033                           reinterpret_cast<const char*>(&mreq), sizeof(mreq));
1034       if (rv)
1035         return MapSystemError(WSAGetLastError());
1036       return OK;
1037     }
1038     default:
1039       NOTREACHED() << "Invalid address family";
1040       return ERR_ADDRESS_INVALID;
1041   }
1042 }
1043
1044 int UDPSocketWin::SetMulticastInterface(uint32 interface_index) {
1045   DCHECK(CalledOnValidThread());
1046   if (is_connected())
1047     return ERR_SOCKET_IS_CONNECTED;
1048   multicast_interface_ = interface_index;
1049   return OK;
1050 }
1051
1052 int UDPSocketWin::SetMulticastTimeToLive(int time_to_live) {
1053   DCHECK(CalledOnValidThread());
1054   if (is_connected())
1055     return ERR_SOCKET_IS_CONNECTED;
1056
1057   if (time_to_live < 0 || time_to_live > 255)
1058     return ERR_INVALID_ARGUMENT;
1059   multicast_time_to_live_ = time_to_live;
1060   return OK;
1061 }
1062
1063 int UDPSocketWin::SetMulticastLoopbackMode(bool loopback) {
1064   DCHECK(CalledOnValidThread());
1065   if (is_connected())
1066     return ERR_SOCKET_IS_CONNECTED;
1067
1068   if (loopback)
1069     socket_options_ |= SOCKET_OPTION_MULTICAST_LOOP;
1070   else
1071     socket_options_ &= ~SOCKET_OPTION_MULTICAST_LOOP;
1072   return OK;
1073 }
1074
1075 int UDPSocketWin::SetDiffServCodePoint(DiffServCodePoint dscp) {
1076   if (dscp == DSCP_NO_CHANGE) {
1077     return OK;
1078   }
1079
1080   if (!is_connected())
1081     return ERR_SOCKET_NOT_CONNECTED;
1082
1083   QwaveAPI& qos(QwaveAPI::Get());
1084
1085   if (!qos.qwave_supported())
1086     return ERROR_NOT_SUPPORTED;
1087
1088   if (qos_handle_ == NULL) {
1089     QOS_VERSION version;
1090     version.MajorVersion = 1;
1091     version.MinorVersion = 0;
1092     qos.CreateHandle(&version, &qos_handle_);
1093     if (qos_handle_ == NULL)
1094       return ERROR_NOT_SUPPORTED;
1095   }
1096
1097   QOS_TRAFFIC_TYPE traffic_type = QOSTrafficTypeBestEffort;
1098   switch (dscp) {
1099     case DSCP_CS0:
1100       traffic_type = QOSTrafficTypeBestEffort;
1101       break;
1102     case DSCP_CS1:
1103       traffic_type = QOSTrafficTypeBackground;
1104       break;
1105     case DSCP_AF11:
1106     case DSCP_AF12:
1107     case DSCP_AF13:
1108     case DSCP_CS2:
1109     case DSCP_AF21:
1110     case DSCP_AF22:
1111     case DSCP_AF23:
1112     case DSCP_CS3:
1113     case DSCP_AF31:
1114     case DSCP_AF32:
1115     case DSCP_AF33:
1116     case DSCP_CS4:
1117       traffic_type = QOSTrafficTypeExcellentEffort;
1118       break;
1119     case DSCP_AF41:
1120     case DSCP_AF42:
1121     case DSCP_AF43:
1122     case DSCP_CS5:
1123       traffic_type = QOSTrafficTypeAudioVideo;
1124       break;
1125     case DSCP_EF:
1126     case DSCP_CS6:
1127       traffic_type = QOSTrafficTypeVoice;
1128       break;
1129     case DSCP_CS7:
1130       traffic_type = QOSTrafficTypeControl;
1131       break;
1132     case DSCP_NO_CHANGE:
1133       NOTREACHED();
1134       break;
1135   }
1136   if (qos_flow_id_ != 0) {
1137     qos.RemoveSocketFromFlow(qos_handle_, NULL, qos_flow_id_, 0);
1138     qos_flow_id_ = 0;
1139   }
1140   if (!qos.AddSocketToFlow(qos_handle_,
1141                            socket_,
1142                            NULL,
1143                            traffic_type,
1144                            QOS_NON_ADAPTIVE_FLOW,
1145                            &qos_flow_id_)) {
1146     DWORD err = GetLastError();
1147     if (err == ERROR_DEVICE_REINITIALIZATION_NEEDED) {
1148       qos.CloseHandle(qos_handle_);
1149       qos_flow_id_ = 0;
1150       qos_handle_ = 0;
1151     }
1152     return MapSystemError(err);
1153   }
1154   // This requires admin rights, and may fail, if so we ignore it
1155   // as AddSocketToFlow should still do *approximately* the right thing.
1156   DWORD buf = dscp;
1157   qos.SetFlow(qos_handle_,
1158               qos_flow_id_,
1159               QOSSetOutgoingDSCPValue,
1160               sizeof(buf),
1161               &buf,
1162               0,
1163               NULL);
1164
1165   return OK;
1166 }
1167
1168 void UDPSocketWin::DetachFromThread() {
1169   base::NonThreadSafe::DetachFromThread();
1170 }
1171
1172 void UDPSocketWin::UseNonBlockingIO() {
1173   DCHECK(!core_);
1174   use_non_blocking_io_ = true;
1175 }
1176
1177 }  // namespace net