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