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