net/udp/udp_socket_libevent.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_libevent.h"
   6
   7 #include <errno.h>
   8 #include <fcntl.h>
   9 #include <netdb.h>
  10 #include <net/if.h>
  11 #include <netinet/in.h>
  12 #include <sys/ioctl.h>
  13 #include <sys/socket.h>
  14
  15 #include "base/callback.h"
  16 #include "base/logging.h"
  17 #include "base/message_loop/message_loop.h"
  18 #include "base/metrics/stats_counters.h"
  19 #include "base/posix/eintr_wrapper.h"
  20 #include "base/rand_util.h"
  21 #include "net/base/io_buffer.h"
  22 #include "net/base/ip_endpoint.h"
  23 #include "net/base/net_errors.h"
  24 #include "net/base/net_log.h"
  25 #include "net/base/net_util.h"
  26 #include "net/socket/socket_descriptor.h"
  27 #include "net/udp/udp_net_log_parameters.h"
  28
  29
  30 namespace net {
  31
  32 namespace {
  33
  34 const int kBindRetries = 10;
  35 const int kPortStart = 1024;
  36 const int kPortEnd = 65535;
  37
  38 #if defined(OS_MACOSX)
  39
  40 // Returns IPv4 address in network order.
  41 int GetIPv4AddressFromIndex(int socket, uint32 index, uint32* address){
  42   if (!index) {
  43     *address = htonl(INADDR_ANY);
  44     return OK;
  45   }
  46   ifreq ifr;
  47   ifr.ifr_addr.sa_family = AF_INET;
  48   if (!if_indextoname(index, ifr.ifr_name))
  49     return ERR_FAILED;
  50   int rv = ioctl(socket, SIOCGIFADDR, &ifr);
  51   if (!rv)
  52     return MapSystemError(rv);
  53   *address = reinterpret_cast<sockaddr_in*>(&ifr.ifr_addr)->sin_addr.s_addr;
  54   return OK;
  55 }
  56
  57 #endif  // OS_MACOSX
  58
  59 }  // namespace
  60
  61 UDPSocketLibevent::UDPSocketLibevent(
  62     DatagramSocket::BindType bind_type,
  63     const RandIntCallback& rand_int_cb,
  64     net::NetLog* net_log,
  65     const net::NetLog::Source& source)
  66         : socket_(kInvalidSocket),
  67           addr_family_(0),
  68           socket_options_(SOCKET_OPTION_MULTICAST_LOOP),
  69           multicast_interface_(0),
  70           multicast_time_to_live_(1),
  71           bind_type_(bind_type),
  72           rand_int_cb_(rand_int_cb),
  73           read_watcher_(this),
  74           write_watcher_(this),
  75           read_buf_len_(0),
  76           recv_from_address_(NULL),
  77           write_buf_len_(0),
  78           net_log_(BoundNetLog::Make(net_log, NetLog::SOURCE_UDP_SOCKET)) {
  79   net_log_.BeginEvent(NetLog::TYPE_SOCKET_ALIVE,
  80                       source.ToEventParametersCallback());
  81   if (bind_type == DatagramSocket::RANDOM_BIND)
  82     DCHECK(!rand_int_cb.is_null());
  83 }
  84
  85 UDPSocketLibevent::~UDPSocketLibevent() {
  86   Close();
  87   net_log_.EndEvent(NetLog::TYPE_SOCKET_ALIVE);
  88 }
  89
  90 void UDPSocketLibevent::Close() {
  91   DCHECK(CalledOnValidThread());
  92
  93   if (!is_connected())
  94     return;
  95
  96   // Zero out any pending read/write callback state.
  97   read_buf_ = NULL;
  98   read_buf_len_ = 0;
  99   read_callback_.Reset();
 100   recv_from_address_ = NULL;
 101   write_buf_ = NULL;
 102   write_buf_len_ = 0;
 103   write_callback_.Reset();
 104   send_to_address_.reset();
 105
 106   bool ok = read_socket_watcher_.StopWatchingFileDescriptor();
 107   DCHECK(ok);
 108   ok = write_socket_watcher_.StopWatchingFileDescriptor();
 109   DCHECK(ok);
 110
 111   if (IGNORE_EINTR(close(socket_)) < 0)
 112     PLOG(ERROR) << "close";
 113
 114   socket_ = kInvalidSocket;
 115   addr_family_ = 0;
 116 }
 117
 118 int UDPSocketLibevent::GetPeerAddress(IPEndPoint* address) const {
 119   DCHECK(CalledOnValidThread());
 120   DCHECK(address);
 121   if (!is_connected())
 122     return ERR_SOCKET_NOT_CONNECTED;
 123
 124   if (!remote_address_.get()) {
 125     SockaddrStorage storage;
 126     if (getpeername(socket_, storage.addr, &storage.addr_len))
 127       return MapSystemError(errno);
 128     scoped_ptr<IPEndPoint> address(new IPEndPoint());
 129     if (!address->FromSockAddr(storage.addr, storage.addr_len))
 130       return ERR_FAILED;
 131     remote_address_.reset(address.release());
 132   }
 133
 134   *address = *remote_address_;
 135   return OK;
 136 }
 137
 138 int UDPSocketLibevent::GetLocalAddress(IPEndPoint* address) const {
 139   DCHECK(CalledOnValidThread());
 140   DCHECK(address);
 141   if (!is_connected())
 142     return ERR_SOCKET_NOT_CONNECTED;
 143
 144   if (!local_address_.get()) {
 145     SockaddrStorage storage;
 146     if (getsockname(socket_, storage.addr, &storage.addr_len))
 147       return MapSystemError(errno);
 148     scoped_ptr<IPEndPoint> address(new IPEndPoint());
 149     if (!address->FromSockAddr(storage.addr, storage.addr_len))
 150       return ERR_FAILED;
 151     local_address_.reset(address.release());
 152     net_log_.AddEvent(NetLog::TYPE_UDP_LOCAL_ADDRESS,
 153                       CreateNetLogUDPConnectCallback(local_address_.get()));
 154   }
 155
 156   *address = *local_address_;
 157   return OK;
 158 }
 159
 160 int UDPSocketLibevent::Read(IOBuffer* buf,
 161                             int buf_len,
 162                             const CompletionCallback& callback) {
 163   return RecvFrom(buf, buf_len, NULL, callback);
 164 }
 165
 166 int UDPSocketLibevent::RecvFrom(IOBuffer* buf,
 167                                 int buf_len,
 168                                 IPEndPoint* address,
 169                                 const CompletionCallback& callback) {
 170   DCHECK(CalledOnValidThread());
 171   DCHECK_NE(kInvalidSocket, socket_);
 172   DCHECK(read_callback_.is_null());
 173   DCHECK(!recv_from_address_);
 174   DCHECK(!callback.is_null());  // Synchronous operation not supported
 175   DCHECK_GT(buf_len, 0);
 176
 177   int nread = InternalRecvFrom(buf, buf_len, address);
 178   if (nread != ERR_IO_PENDING)
 179     return nread;
 180
 181   if (!base::MessageLoopForIO::current()->WatchFileDescriptor(
 182           socket_, true, base::MessageLoopForIO::WATCH_READ,
 183           &read_socket_watcher_, &read_watcher_)) {
 184     PLOG(ERROR) << "WatchFileDescriptor failed on read";
 185     int result = MapSystemError(errno);
 186     LogRead(result, NULL, 0, NULL);
 187     return result;
 188   }
 189
 190   read_buf_ = buf;
 191   read_buf_len_ = buf_len;
 192   recv_from_address_ = address;
 193   read_callback_ = callback;
 194   return ERR_IO_PENDING;
 195 }
 196
 197 int UDPSocketLibevent::Write(IOBuffer* buf,
 198                              int buf_len,
 199                              const CompletionCallback& callback) {
 200   return SendToOrWrite(buf, buf_len, NULL, callback);
 201 }
 202
 203 int UDPSocketLibevent::SendTo(IOBuffer* buf,
 204                               int buf_len,
 205                               const IPEndPoint& address,
 206                               const CompletionCallback& callback) {
 207   return SendToOrWrite(buf, buf_len, &address, callback);
 208 }
 209
 210 int UDPSocketLibevent::SendToOrWrite(IOBuffer* buf,
 211                                      int buf_len,
 212                                      const IPEndPoint* address,
 213                                      const CompletionCallback& callback) {
 214   DCHECK(CalledOnValidThread());
 215   DCHECK_NE(kInvalidSocket, socket_);
 216   DCHECK(write_callback_.is_null());
 217   DCHECK(!callback.is_null());  // Synchronous operation not supported
 218   DCHECK_GT(buf_len, 0);
 219
 220   int result = InternalSendTo(buf, buf_len, address);
 221   if (result != ERR_IO_PENDING)
 222     return result;
 223
 224   if (!base::MessageLoopForIO::current()->WatchFileDescriptor(
 225           socket_, true, base::MessageLoopForIO::WATCH_WRITE,
 226           &write_socket_watcher_, &write_watcher_)) {
 227     DVLOG(1) << "WatchFileDescriptor failed on write, errno " << errno;
 228     int result = MapSystemError(errno);
 229     LogWrite(result, NULL, NULL);
 230     return result;
 231   }
 232
 233   write_buf_ = buf;
 234   write_buf_len_ = buf_len;
 235   DCHECK(!send_to_address_.get());
 236   if (address) {
 237     send_to_address_.reset(new IPEndPoint(*address));
 238   }
 239   write_callback_ = callback;
 240   return ERR_IO_PENDING;
 241 }
 242
 243 int UDPSocketLibevent::Connect(const IPEndPoint& address) {
 244   net_log_.BeginEvent(NetLog::TYPE_UDP_CONNECT,
 245                       CreateNetLogUDPConnectCallback(&address));
 246   int rv = InternalConnect(address);
 247   if (rv != OK)
 248     Close();
 249   net_log_.EndEventWithNetErrorCode(NetLog::TYPE_UDP_CONNECT, rv);
 250   return rv;
 251 }
 252
 253 int UDPSocketLibevent::InternalConnect(const IPEndPoint& address) {
 254   DCHECK(CalledOnValidThread());
 255   DCHECK(!is_connected());
 256   DCHECK(!remote_address_.get());
 257   int rv = CreateSocket(address);
 258   if (rv < 0)
 259     return rv;
 260
 261   if (bind_type_ == DatagramSocket::RANDOM_BIND)
 262     rv = RandomBind(address);
 263   // else connect() does the DatagramSocket::DEFAULT_BIND
 264
 265   if (rv < 0) {
 266     Close();
 267     return rv;
 268   }
 269
 270   SockaddrStorage storage;
 271   if (!address.ToSockAddr(storage.addr, &storage.addr_len)) {
 272     Close();
 273     return ERR_ADDRESS_INVALID;
 274   }
 275
 276   rv = HANDLE_EINTR(connect(socket_, storage.addr, storage.addr_len));
 277   if (rv < 0) {
 278     // Close() may change the current errno. Map errno beforehand.
 279     int result = MapSystemError(errno);
 280     Close();
 281     return result;
 282   }
 283
 284   remote_address_.reset(new IPEndPoint(address));
 285   return rv;
 286 }
 287
 288 int UDPSocketLibevent::Bind(const IPEndPoint& address) {
 289   DCHECK(CalledOnValidThread());
 290   DCHECK(!is_connected());
 291   int rv = CreateSocket(address);
 292   if (rv < 0)
 293     return rv;
 294
 295   rv = SetSocketOptions();
 296   if (rv < 0) {
 297     Close();
 298     return rv;
 299   }
 300   rv = DoBind(address);
 301   if (rv < 0) {
 302     Close();
 303     return rv;
 304   }
 305   local_address_.reset();
 306   return rv;
 307 }
 308
 309 bool UDPSocketLibevent::SetReceiveBufferSize(int32 size) {
 310   DCHECK(CalledOnValidThread());
 311   int rv = setsockopt(socket_, SOL_SOCKET, SO_RCVBUF,
 312                       reinterpret_cast<const char*>(&size), sizeof(size));
 313   DCHECK(!rv) << "Could not set socket receive buffer size: " << errno;
 314   return rv == 0;
 315 }
 316
 317 bool UDPSocketLibevent::SetSendBufferSize(int32 size) {
 318   DCHECK(CalledOnValidThread());
 319   int rv = setsockopt(socket_, SOL_SOCKET, SO_SNDBUF,
 320                       reinterpret_cast<const char*>(&size), sizeof(size));
 321   DCHECK(!rv) << "Could not set socket send buffer size: " << errno;
 322   return rv == 0;
 323 }
 324
 325 void UDPSocketLibevent::AllowAddressReuse() {
 326   DCHECK(CalledOnValidThread());
 327   DCHECK(!is_connected());
 328
 329   socket_options_ |= SOCKET_OPTION_REUSE_ADDRESS;
 330 }
 331
 332 void UDPSocketLibevent::AllowBroadcast() {
 333   DCHECK(CalledOnValidThread());
 334   DCHECK(!is_connected());
 335
 336   socket_options_ |= SOCKET_OPTION_BROADCAST;
 337 }
 338
 339 void UDPSocketLibevent::ReadWatcher::OnFileCanReadWithoutBlocking(int) {
 340   if (!socket_->read_callback_.is_null())
 341     socket_->DidCompleteRead();
 342 }
 343
 344 void UDPSocketLibevent::WriteWatcher::OnFileCanWriteWithoutBlocking(int) {
 345   if (!socket_->write_callback_.is_null())
 346     socket_->DidCompleteWrite();
 347 }
 348
 349 void UDPSocketLibevent::DoReadCallback(int rv) {
 350   DCHECK_NE(rv, ERR_IO_PENDING);
 351   DCHECK(!read_callback_.is_null());
 352
 353   // since Run may result in Read being called, clear read_callback_ up front.
 354   CompletionCallback c = read_callback_;
 355   read_callback_.Reset();
 356   c.Run(rv);
 357 }
 358
 359 void UDPSocketLibevent::DoWriteCallback(int rv) {
 360   DCHECK_NE(rv, ERR_IO_PENDING);
 361   DCHECK(!write_callback_.is_null());
 362
 363   // since Run may result in Write being called, clear write_callback_ up front.
 364   CompletionCallback c = write_callback_;
 365   write_callback_.Reset();
 366   c.Run(rv);
 367 }
 368
 369 void UDPSocketLibevent::DidCompleteRead() {
 370   int result =
 371       InternalRecvFrom(read_buf_.get(), read_buf_len_, recv_from_address_);
 372   if (result != ERR_IO_PENDING) {
 373     read_buf_ = NULL;
 374     read_buf_len_ = 0;
 375     recv_from_address_ = NULL;
 376     bool ok = read_socket_watcher_.StopWatchingFileDescriptor();
 377     DCHECK(ok);
 378     DoReadCallback(result);
 379   }
 380 }
 381
 382 void UDPSocketLibevent::LogRead(int result,
 383                                 const char* bytes,
 384                                 socklen_t addr_len,
 385                                 const sockaddr* addr) const {
 386   if (result < 0) {
 387     net_log_.AddEventWithNetErrorCode(NetLog::TYPE_UDP_RECEIVE_ERROR, result);
 388     return;
 389   }
 390
 391   if (net_log_.IsLoggingAllEvents()) {
 392     DCHECK(addr_len > 0);
 393     DCHECK(addr);
 394
 395     IPEndPoint address;
 396     bool is_address_valid = address.FromSockAddr(addr, addr_len);
 397     net_log_.AddEvent(
 398         NetLog::TYPE_UDP_BYTES_RECEIVED,
 399         CreateNetLogUDPDataTranferCallback(
 400             result, bytes,
 401             is_address_valid ? &address : NULL));
 402   }
 403
 404   base::StatsCounter read_bytes("udp.read_bytes");
 405   read_bytes.Add(result);
 406 }
 407
 408 int UDPSocketLibevent::CreateSocket(const IPEndPoint& address) {
 409   addr_family_ = address.GetSockAddrFamily();
 410   socket_ = CreatePlatformSocket(addr_family_, SOCK_DGRAM, 0);
 411   if (socket_ == kInvalidSocket)
 412     return MapSystemError(errno);
 413   if (SetNonBlocking(socket_)) {
 414     const int err = MapSystemError(errno);
 415     Close();
 416     return err;
 417   }
 418   return OK;
 419 }
 420
 421 void UDPSocketLibevent::DidCompleteWrite() {
 422   int result =
 423       InternalSendTo(write_buf_.get(), write_buf_len_, send_to_address_.get());
 424
 425   if (result != ERR_IO_PENDING) {
 426     write_buf_ = NULL;
 427     write_buf_len_ = 0;
 428     send_to_address_.reset();
 429     write_socket_watcher_.StopWatchingFileDescriptor();
 430     DoWriteCallback(result);
 431   }
 432 }
 433
 434 void UDPSocketLibevent::LogWrite(int result,
 435                                  const char* bytes,
 436                                  const IPEndPoint* address) const {
 437   if (result < 0) {
 438     net_log_.AddEventWithNetErrorCode(NetLog::TYPE_UDP_SEND_ERROR, result);
 439     return;
 440   }
 441
 442   if (net_log_.IsLoggingAllEvents()) {
 443     net_log_.AddEvent(
 444         NetLog::TYPE_UDP_BYTES_SENT,
 445         CreateNetLogUDPDataTranferCallback(result, bytes, address));
 446   }
 447
 448   base::StatsCounter write_bytes("udp.write_bytes");
 449   write_bytes.Add(result);
 450 }
 451
 452 int UDPSocketLibevent::InternalRecvFrom(IOBuffer* buf, int buf_len,
 453                                         IPEndPoint* address) {
 454   int bytes_transferred;
 455   int flags = 0;
 456
 457   SockaddrStorage storage;
 458
 459   bytes_transferred =
 460       HANDLE_EINTR(recvfrom(socket_,
 461                             buf->data(),
 462                             buf_len,
 463                             flags,
 464                             storage.addr,
 465                             &storage.addr_len));
 466   int result;
 467   if (bytes_transferred >= 0) {
 468     result = bytes_transferred;
 469     if (address && !address->FromSockAddr(storage.addr, storage.addr_len))
 470       result = ERR_FAILED;
 471   } else {
 472     result = MapSystemError(errno);
 473   }
 474   if (result != ERR_IO_PENDING)
 475     LogRead(result, buf->data(), storage.addr_len, storage.addr);
 476   return result;
 477 }
 478
 479 int UDPSocketLibevent::InternalSendTo(IOBuffer* buf, int buf_len,
 480                                       const IPEndPoint* address) {
 481   SockaddrStorage storage;
 482   struct sockaddr* addr = storage.addr;
 483   if (!address) {
 484     addr = NULL;
 485     storage.addr_len = 0;
 486   } else {
 487     if (!address->ToSockAddr(storage.addr, &storage.addr_len)) {
 488       int result = ERR_FAILED;
 489       LogWrite(result, NULL, NULL);
 490       return result;
 491     }
 492   }
 493
 494   int result = HANDLE_EINTR(sendto(socket_,
 495                             buf->data(),
 496                             buf_len,
 497                             0,
 498                             addr,
 499                             storage.addr_len));
 500   if (result < 0)
 501     result = MapSystemError(errno);
 502   if (result != ERR_IO_PENDING)
 503     LogWrite(result, buf->data(), address);
 504   return result;
 505 }
 506
 507 int UDPSocketLibevent::SetSocketOptions() {
 508   int true_value = 1;
 509   if (socket_options_ & SOCKET_OPTION_REUSE_ADDRESS) {
 510     int rv = setsockopt(socket_, SOL_SOCKET, SO_REUSEADDR, &true_value,
 511                         sizeof(true_value));
 512     if (rv < 0)
 513       return MapSystemError(errno);
 514   }
 515   if (socket_options_ & SOCKET_OPTION_BROADCAST) {
 516     int rv;
 517 #if defined(OS_MACOSX)
 518     // SO_REUSEPORT on OSX permits multiple processes to each receive
 519     // UDP multicast or broadcast datagrams destined for the bound
 520     // port.
 521     rv = setsockopt(socket_, SOL_SOCKET, SO_REUSEPORT, &true_value,
 522                     sizeof(true_value));
 523 #else
 524     rv = setsockopt(socket_, SOL_SOCKET, SO_BROADCAST, &true_value,
 525                     sizeof(true_value));
 526 #endif  // defined(OS_MACOSX)
 527     if (rv < 0)
 528       return MapSystemError(errno);
 529   }
 530
 531   if (!(socket_options_ & SOCKET_OPTION_MULTICAST_LOOP)) {
 532     int rv;
 533     if (addr_family_ == AF_INET) {
 534       u_char loop = 0;
 535       rv = setsockopt(socket_, IPPROTO_IP, IP_MULTICAST_LOOP,
 536                       &loop, sizeof(loop));
 537     } else {
 538       u_int loop = 0;
 539       rv = setsockopt(socket_, IPPROTO_IPV6, IPV6_MULTICAST_LOOP,
 540                       &loop, sizeof(loop));
 541     }
 542     if (rv < 0)
 543       return MapSystemError(errno);
 544   }
 545   if (multicast_time_to_live_ != IP_DEFAULT_MULTICAST_TTL) {
 546     int rv;
 547     if (addr_family_ == AF_INET) {
 548       u_char ttl = multicast_time_to_live_;
 549       rv = setsockopt(socket_, IPPROTO_IP, IP_MULTICAST_TTL,
 550                       &ttl, sizeof(ttl));
 551     } else {
 552       // Signed integer. -1 to use route default.
 553       int ttl = multicast_time_to_live_;
 554       rv = setsockopt(socket_, IPPROTO_IPV6, IPV6_MULTICAST_HOPS,
 555                       &ttl, sizeof(ttl));
 556     }
 557     if (rv < 0)
 558       return MapSystemError(errno);
 559   }
 560   if (multicast_interface_ != 0) {
 561     switch (addr_family_) {
 562       case AF_INET: {
 563 #if !defined(OS_MACOSX)
 564         ip_mreqn mreq;
 565         mreq.imr_ifindex = multicast_interface_;
 566         mreq.imr_address.s_addr = htonl(INADDR_ANY);
 567 #else
 568         ip_mreq mreq;
 569         int error = GetIPv4AddressFromIndex(socket_, multicast_interface_,
 570                                             &mreq.imr_interface.s_addr);
 571         if (error != OK)
 572           return error;
 573 #endif
 574         int rv = setsockopt(socket_, IPPROTO_IP, IP_MULTICAST_IF,
 575                             reinterpret_cast<const char*>(&mreq), sizeof(mreq));
 576         if (rv)
 577           return MapSystemError(errno);
 578         break;
 579       }
 580       case AF_INET6: {
 581         uint32 interface_index = multicast_interface_;
 582         int rv = setsockopt(socket_, IPPROTO_IPV6, IPV6_MULTICAST_IF,
 583                             reinterpret_cast<const char*>(&interface_index),
 584                             sizeof(interface_index));
 585         if (rv)
 586           return MapSystemError(errno);
 587         break;
 588       }
 589       default:
 590         NOTREACHED() << "Invalid address family";
 591         return ERR_ADDRESS_INVALID;
 592     }
 593   }
 594   return OK;
 595 }
 596
 597 int UDPSocketLibevent::DoBind(const IPEndPoint& address) {
 598   SockaddrStorage storage;
 599   if (!address.ToSockAddr(storage.addr, &storage.addr_len))
 600     return ERR_ADDRESS_INVALID;
 601   int rv = bind(socket_, storage.addr, storage.addr_len);
 602   return rv < 0 ? MapSystemError(errno) : rv;
 603 }
 604
 605 int UDPSocketLibevent::RandomBind(const IPEndPoint& address) {
 606   DCHECK(bind_type_ == DatagramSocket::RANDOM_BIND && !rand_int_cb_.is_null());
 607
 608   // Construct IPAddressNumber of appropriate size (IPv4 or IPv6) of 0s.
 609   IPAddressNumber ip(address.address().size());
 610
 611   for (int i = 0; i < kBindRetries; ++i) {
 612     int rv = DoBind(IPEndPoint(ip, rand_int_cb_.Run(kPortStart, kPortEnd)));
 613     if (rv == OK || rv != ERR_ADDRESS_IN_USE)
 614       return rv;
 615   }
 616   return DoBind(IPEndPoint(ip, 0));
 617 }
 618
 619 int UDPSocketLibevent::JoinGroup(const IPAddressNumber& group_address) const {
 620   DCHECK(CalledOnValidThread());
 621   if (!is_connected())
 622     return ERR_SOCKET_NOT_CONNECTED;
 623
 624   switch (group_address.size()) {
 625     case kIPv4AddressSize: {
 626       if (addr_family_ != AF_INET)
 627         return ERR_ADDRESS_INVALID;
 628
 629 #if !defined(OS_MACOSX)
 630       ip_mreqn mreq;
 631       mreq.imr_ifindex = multicast_interface_;
 632       mreq.imr_address.s_addr = htonl(INADDR_ANY);
 633 #else
 634       ip_mreq mreq;
 635       int error = GetIPv4AddressFromIndex(socket_, multicast_interface_,
 636                                             &mreq.imr_interface.s_addr);
 637       if (error != OK)
 638         return error;
 639 #endif
 640       memcpy(&mreq.imr_multiaddr, &group_address[0], kIPv4AddressSize);
 641       int rv = setsockopt(socket_, IPPROTO_IP, IP_ADD_MEMBERSHIP,
 642                           &mreq, sizeof(mreq));
 643       if (rv < 0)
 644         return MapSystemError(errno);
 645       return OK;
 646     }
 647     case kIPv6AddressSize: {
 648       if (addr_family_ != AF_INET6)
 649         return ERR_ADDRESS_INVALID;
 650       ipv6_mreq mreq;
 651       mreq.ipv6mr_interface = multicast_interface_;
 652       memcpy(&mreq.ipv6mr_multiaddr, &group_address[0], kIPv6AddressSize);
 653       int rv = setsockopt(socket_, IPPROTO_IPV6, IPV6_JOIN_GROUP,
 654                           &mreq, sizeof(mreq));
 655       if (rv < 0)
 656         return MapSystemError(errno);
 657       return OK;
 658     }
 659     default:
 660       NOTREACHED() << "Invalid address family";
 661       return ERR_ADDRESS_INVALID;
 662   }
 663 }
 664
 665 int UDPSocketLibevent::LeaveGroup(const IPAddressNumber& group_address) const {
 666   DCHECK(CalledOnValidThread());
 667
 668   if (!is_connected())
 669     return ERR_SOCKET_NOT_CONNECTED;
 670
 671   switch (group_address.size()) {
 672     case kIPv4AddressSize: {
 673       if (addr_family_ != AF_INET)
 674         return ERR_ADDRESS_INVALID;
 675       ip_mreq mreq;
 676       mreq.imr_interface.s_addr = INADDR_ANY;
 677       memcpy(&mreq.imr_multiaddr, &group_address[0], kIPv4AddressSize);
 678       int rv = setsockopt(socket_, IPPROTO_IP, IP_DROP_MEMBERSHIP,
 679                           &mreq, sizeof(mreq));
 680       if (rv < 0)
 681         return MapSystemError(errno);
 682       return OK;
 683     }
 684     case kIPv6AddressSize: {
 685       if (addr_family_ != AF_INET6)
 686         return ERR_ADDRESS_INVALID;
 687       ipv6_mreq mreq;
 688       mreq.ipv6mr_interface = 0;  // 0 indicates default multicast interface.
 689       memcpy(&mreq.ipv6mr_multiaddr, &group_address[0], kIPv6AddressSize);
 690       int rv = setsockopt(socket_, IPPROTO_IPV6, IPV6_LEAVE_GROUP,
 691                           &mreq, sizeof(mreq));
 692       if (rv < 0)
 693         return MapSystemError(errno);
 694       return OK;
 695     }
 696     default:
 697       NOTREACHED() << "Invalid address family";
 698       return ERR_ADDRESS_INVALID;
 699   }
 700 }
 701
 702 int UDPSocketLibevent::SetMulticastInterface(uint32 interface_index) {
 703   DCHECK(CalledOnValidThread());
 704   if (is_connected())
 705     return ERR_SOCKET_IS_CONNECTED;
 706   multicast_interface_ = interface_index;
 707   return OK;
 708 }
 709
 710 int UDPSocketLibevent::SetMulticastTimeToLive(int time_to_live) {
 711   DCHECK(CalledOnValidThread());
 712   if (is_connected())
 713     return ERR_SOCKET_IS_CONNECTED;
 714
 715   if (time_to_live < 0 || time_to_live > 255)
 716     return ERR_INVALID_ARGUMENT;
 717   multicast_time_to_live_ = time_to_live;
 718   return OK;
 719 }
 720
 721 int UDPSocketLibevent::SetMulticastLoopbackMode(bool loopback) {
 722   DCHECK(CalledOnValidThread());
 723   if (is_connected())
 724     return ERR_SOCKET_IS_CONNECTED;
 725
 726   if (loopback)
 727     socket_options_ |= SOCKET_OPTION_MULTICAST_LOOP;
 728   else
 729     socket_options_ &= ~SOCKET_OPTION_MULTICAST_LOOP;
 730   return OK;
 731 }
 732
 733 int UDPSocketLibevent::SetDiffServCodePoint(DiffServCodePoint dscp) {
 734   if (dscp == DSCP_NO_CHANGE) {
 735     return OK;
 736   }
 737   int rv;
 738   int dscp_and_ecn = dscp << 2;
 739   if (addr_family_ == AF_INET) {
 740     rv = setsockopt(socket_, IPPROTO_IP, IP_TOS,
 741                     &dscp_and_ecn, sizeof(dscp_and_ecn));
 742   } else {
 743     rv = setsockopt(socket_, IPPROTO_IPV6, IPV6_TCLASS,
 744                     &dscp_and_ecn, sizeof(dscp_and_ecn));
 745   }
 746   if (rv < 0)
 747     return MapSystemError(errno);
 748
 749   return OK;
 750 }
 751
 752 }  // namespace net