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