text: // Copyright (c) 2012 The Chromium Authors. All rights reserved.
  // Use of this source code is governed by a BSD-style license that can be
  // found in the LICENSE file.
  
  #include "net/udp/udp_socket_libevent.h"
  
  #include <errno.h>
  #include <fcntl.h>
  #include <netdb.h>
  #include <sys/socket.h>
  
  #include "base/callback.h"
  #include "base/logging.h"
  #include "base/message_loop.h"
  #include "base/metrics/stats_counters.h"
  #include "base/posix/eintr_wrapper.h"
  #include "base/rand_util.h"
  #include "net/base/io_buffer.h"
  #include "net/base/ip_endpoint.h"
  #include "net/base/net_errors.h"
  #include "net/base/net_log.h"
  #include "net/base/net_util.h"
  #include "net/udp/udp_net_log_parameters.h"
  #if defined(OS_POSIX)
  #include <netinet/in.h>
  #endif
  
  namespace {
  
  static const int kBindRetries = 10;
  static const int kPortStart = 1024;
  static const int kPortEnd = 65535;
  
  }  // namespace net
  
  namespace net {
  
  UDPSocketLibevent::UDPSocketLibevent(
      DatagramSocket::BindType bind_type,
      const RandIntCallback& rand_int_cb,
      net::NetLog* net_log,
      const net::NetLog::Source& source)
          : socket_(kInvalidSocket),
            addr_family_(0),
            socket_options_(SOCKET_OPTION_MULTICAST_LOOP),
            multicast_time_to_live_(1),
            bind_type_(bind_type),
            rand_int_cb_(rand_int_cb),
            read_watcher_(this),
            write_watcher_(this),
            read_buf_len_(0),
            recv_from_address_(NULL),
            write_buf_len_(0),
            net_log_(BoundNetLog::Make(net_log, NetLog::SOURCE_UDP_SOCKET)) {
    net_log_.BeginEvent(NetLog::TYPE_SOCKET_ALIVE,
                        source.ToEventParametersCallback());
    if (bind_type == DatagramSocket::RANDOM_BIND)
      DCHECK(!rand_int_cb.is_null());
  }
  
  UDPSocketLibevent::~UDPSocketLibevent() {
    Close();
    net_log_.EndEvent(NetLog::TYPE_SOCKET_ALIVE);
  }
  
  void UDPSocketLibevent::Close() {
    DCHECK(CalledOnValidThread());
  
    if (!is_connected())
      return;
  
    // Zero out any pending read/write callback state.
    read_buf_ = NULL;
    read_buf_len_ = 0;
    read_callback_.Reset();
    recv_from_address_ = NULL;
    write_buf_ = NULL;
    write_buf_len_ = 0;
    write_callback_.Reset();
    send_to_address_.reset();
  
    bool ok = read_socket_watcher_.StopWatchingFileDescriptor();
    DCHECK(ok);
    ok = write_socket_watcher_.StopWatchingFileDescriptor();
    DCHECK(ok);
  
    if (HANDLE_EINTR(close(socket_)) < 0)
      PLOG(ERROR) << "close";
  
    socket_ = kInvalidSocket;
  }
  
  int UDPSocketLibevent::GetPeerAddress(IPEndPoint* address) const {
    DCHECK(CalledOnValidThread());
    DCHECK(address);
    if (!is_connected())
      return ERR_SOCKET_NOT_CONNECTED;
  
    if (!remote_address_.get()) {
      SockaddrStorage storage;
      if (getpeername(socket_, storage.addr, &storage.addr_len))
        return MapSystemError(errno);
      scoped_ptr<IPEndPoint> address(new IPEndPoint());
      if (!address->FromSockAddr(storage.addr, storage.addr_len))
        return ERR_FAILED;
      remote_address_.reset(address.release());
    }
  
    *address = *remote_address_;
    return OK;
  }
  
  int UDPSocketLibevent::GetLocalAddress(IPEndPoint* address) const {
    DCHECK(CalledOnValidThread());
    DCHECK(address);
    if (!is_connected())
      return ERR_SOCKET_NOT_CONNECTED;
  
    if (!local_address_.get()) {
      SockaddrStorage storage;
      if (getsockname(socket_, storage.addr, &storage.addr_len))
        return MapSystemError(errno);
      scoped_ptr<IPEndPoint> address(new IPEndPoint());
      if (!address->FromSockAddr(storage.addr, storage.addr_len))
        return ERR_FAILED;
      local_address_.reset(address.release());
    }
  
    *address = *local_address_;
    return OK;
  }
  
  int UDPSocketLibevent::Read(IOBuffer* buf,
                              int buf_len,
                              const CompletionCallback& callback) {
    return RecvFrom(buf, buf_len, NULL, callback);
  }
  
  int UDPSocketLibevent::RecvFrom(IOBuffer* buf,
                                  int buf_len,
                                  IPEndPoint* address,
                                  const CompletionCallback& callback) {
    DCHECK(CalledOnValidThread());
    DCHECK_NE(kInvalidSocket, socket_);
    DCHECK(read_callback_.is_null());
    DCHECK(!recv_from_address_);
    DCHECK(!callback.is_null());  // Synchronous operation not supported
    DCHECK_GT(buf_len, 0);
  
    int nread = InternalRecvFrom(buf, buf_len, address);
    if (nread != ERR_IO_PENDING)
      return nread;
  
    if (!base::MessageLoopForIO::current()->WatchFileDescriptor(
            socket_, true, base::MessageLoopForIO::WATCH_READ,
            &read_socket_watcher_, &read_watcher_)) {
      PLOG(ERROR) << "WatchFileDescriptor failed on read";
      int result = MapSystemError(errno);
      LogRead(result, NULL, 0, NULL);
      return result;
    }
  
    read_buf_ = buf;
    read_buf_len_ = buf_len;
    recv_from_address_ = address;
    read_callback_ = callback;
    return ERR_IO_PENDING;
  }
  
  int UDPSocketLibevent::Write(IOBuffer* buf,
                               int buf_len,
                               const CompletionCallback& callback) {
    return SendToOrWrite(buf, buf_len, NULL, callback);
  }
  
  int UDPSocketLibevent::SendTo(IOBuffer* buf,
                                int buf_len,
                                const IPEndPoint& address,
                                const CompletionCallback& callback) {
    return SendToOrWrite(buf, buf_len, &address, callback);
  }
  
  int UDPSocketLibevent::SendToOrWrite(IOBuffer* buf,
                                       int buf_len,
                                       const IPEndPoint* address,
                                       const CompletionCallback& callback) {
    DCHECK(CalledOnValidThread());
    DCHECK_NE(kInvalidSocket, socket_);
    DCHECK(write_callback_.is_null());
    DCHECK(!callback.is_null());  // Synchronous operation not supported
    DCHECK_GT(buf_len, 0);
  
    int result = InternalSendTo(buf, buf_len, address);
    if (result != ERR_IO_PENDING)
      return result;
  
    if (!base::MessageLoopForIO::current()->WatchFileDescriptor(
            socket_, true, base::MessageLoopForIO::WATCH_WRITE,
            &write_socket_watcher_, &write_watcher_)) {
      DVLOG(1) << "WatchFileDescriptor failed on write, errno " << errno;
      int result = MapSystemError(errno);
      LogWrite(result, NULL, NULL);
      return result;
    }
  
    write_buf_ = buf;
    write_buf_len_ = buf_len;
    DCHECK(!send_to_address_.get());
    if (address) {
      send_to_address_.reset(new IPEndPoint(*address));
    }
    write_callback_ = callback;
    return ERR_IO_PENDING;
  }
  
  int UDPSocketLibevent::Connect(const IPEndPoint& address) {
    net_log_.BeginEvent(NetLog::TYPE_UDP_CONNECT,
                        CreateNetLogUDPConnectCallback(&address));
    int rv = InternalConnect(address);
    if (rv != OK)
      Close();
    net_log_.EndEventWithNetErrorCode(NetLog::TYPE_UDP_CONNECT, rv);
    return rv;
  }
  
  int UDPSocketLibevent::InternalConnect(const IPEndPoint& address) {
    DCHECK(CalledOnValidThread());
    DCHECK(!is_connected());
    DCHECK(!remote_address_.get());
    int rv = CreateSocket(address);
    if (rv < 0)
      return rv;
  
    if (bind_type_ == DatagramSocket::RANDOM_BIND)
      rv = RandomBind(address);
    // else connect() does the DatagramSocket::DEFAULT_BIND
  
    if (rv < 0)
      return rv;
  
    SockaddrStorage storage;
    if (!address.ToSockAddr(storage.addr, &storage.addr_len))
      return ERR_FAILED;
  
    rv = HANDLE_EINTR(connect(socket_, storage.addr, storage.addr_len));
    if (rv < 0)
      return MapSystemError(errno);
  
    remote_address_.reset(new IPEndPoint(address));
    return rv;
  }
  
  int UDPSocketLibevent::Bind(const IPEndPoint& address) {
    DCHECK(CalledOnValidThread());
    DCHECK(!is_connected());
    int rv = CreateSocket(address);
    if (rv < 0)
      return rv;
    rv = SetSocketOptions();
    if (rv < 0)
      return rv;
    rv = DoBind(address);
    if (rv < 0)
      return rv;
    local_address_.reset();
    return rv;
  }
  
  bool UDPSocketLibevent::SetReceiveBufferSize(int32 size) {
    DCHECK(CalledOnValidThread());
    int rv = setsockopt(socket_, SOL_SOCKET, SO_RCVBUF,
                        reinterpret_cast<const char*>(&size), sizeof(size));
    DCHECK(!rv) << "Could not set socket receive buffer size: " << errno;
    return rv == 0;
  }
  
  bool UDPSocketLibevent::SetSendBufferSize(int32 size) {
    DCHECK(CalledOnValidThread());
    int rv = setsockopt(socket_, SOL_SOCKET, SO_SNDBUF,
                        reinterpret_cast<const char*>(&size), sizeof(size));
    DCHECK(!rv) << "Could not set socket send buffer size: " << errno;
    return rv == 0;
  }
  
  void UDPSocketLibevent::AllowAddressReuse() {
    DCHECK(CalledOnValidThread());
    DCHECK(!is_connected());
  
    socket_options_ |= SOCKET_OPTION_REUSE_ADDRESS;
  }
  
  void UDPSocketLibevent::AllowBroadcast() {
    DCHECK(CalledOnValidThread());
    DCHECK(!is_connected());
  
    socket_options_ |= SOCKET_OPTION_BROADCAST;
  }
  
  void UDPSocketLibevent::ReadWatcher::OnFileCanReadWithoutBlocking(int) {
    if (!socket_->read_callback_.is_null())
      socket_->DidCompleteRead();
  }
  
  void UDPSocketLibevent::WriteWatcher::OnFileCanWriteWithoutBlocking(int) {
    if (!socket_->write_callback_.is_null())
      socket_->DidCompleteWrite();
  }
  
  void UDPSocketLibevent::DoReadCallback(int rv) {
    DCHECK_NE(rv, ERR_IO_PENDING);
    DCHECK(!read_callback_.is_null());
  
    // since Run may result in Read being called, clear read_callback_ up front.
    CompletionCallback c = read_callback_;
    read_callback_.Reset();
    c.Run(rv);
  }
  
  void UDPSocketLibevent::DoWriteCallback(int rv) {
    DCHECK_NE(rv, ERR_IO_PENDING);
    DCHECK(!write_callback_.is_null());
  
    // since Run may result in Write being called, clear write_callback_ up front.
    CompletionCallback c = write_callback_;
    write_callback_.Reset();
    c.Run(rv);
  }
  
  void UDPSocketLibevent::DidCompleteRead() {
    int result =
        InternalRecvFrom(read_buf_.get(), read_buf_len_, recv_from_address_);
    if (result != ERR_IO_PENDING) {
      read_buf_ = NULL;
      read_buf_len_ = 0;
      recv_from_address_ = NULL;
      bool ok = read_socket_watcher_.StopWatchingFileDescriptor();
      DCHECK(ok);
      DoReadCallback(result);
    }
  }
  
  void UDPSocketLibevent::LogRead(int result,
                                  const char* bytes,
                                  socklen_t addr_len,
                                  const sockaddr* addr) const {
    if (result < 0) {
      net_log_.AddEventWithNetErrorCode(NetLog::TYPE_UDP_RECEIVE_ERROR, result);
      return;
    }
  
    if (net_log_.IsLoggingAllEvents()) {
      DCHECK(addr_len > 0);
      DCHECK(addr);
  
      IPEndPoint address;
      bool is_address_valid = address.FromSockAddr(addr, addr_len);
      net_log_.AddEvent(
          NetLog::TYPE_UDP_BYTES_RECEIVED,
          CreateNetLogUDPDataTranferCallback(
              result, bytes,
              is_address_valid ? &address : NULL));
    }
  
    base::StatsCounter read_bytes("udp.read_bytes");
    read_bytes.Add(result);
  }
  
  int UDPSocketLibevent::CreateSocket(const IPEndPoint& address) {
    addr_family_ = address.GetSockAddrFamily();
    socket_ = socket(addr_family_, SOCK_DGRAM, 0);
    if (socket_ == kInvalidSocket)
      return MapSystemError(errno);
    if (SetNonBlocking(socket_)) {
      const int err = MapSystemError(errno);
      Close();
      return err;
    }
    return OK;
  }
  
  void UDPSocketLibevent::DidCompleteWrite() {
    int result =
        InternalSendTo(write_buf_.get(), write_buf_len_, send_to_address_.get());
  
    if (result != ERR_IO_PENDING) {
      write_buf_ = NULL;
      write_buf_len_ = 0;
      send_to_address_.reset();
      write_socket_watcher_.StopWatchingFileDescriptor();
      DoWriteCallback(result);
    }
  }
  
  void UDPSocketLibevent::LogWrite(int result,
                                   const char* bytes,
                                   const IPEndPoint* address) const {
    if (result < 0) {
      net_log_.AddEventWithNetErrorCode(NetLog::TYPE_UDP_SEND_ERROR, result);
      return;
    }
  
    if (net_log_.IsLoggingAllEvents()) {
      net_log_.AddEvent(
          NetLog::TYPE_UDP_BYTES_SENT,
          CreateNetLogUDPDataTranferCallback(result, bytes, address));
    }
  
    base::StatsCounter write_bytes("udp.write_bytes");
    write_bytes.Add(result);
  }
  
  int UDPSocketLibevent::InternalRecvFrom(IOBuffer* buf, int buf_len,
                                          IPEndPoint* address) {
    int bytes_transferred;
    int flags = 0;
  
    SockaddrStorage storage;
  
    bytes_transferred =
        HANDLE_EINTR(recvfrom(socket_,
                              buf->data(),
                              buf_len,
                              flags,
                              storage.addr,
                              &storage.addr_len));
    int result;
    if (bytes_transferred >= 0) {
      result = bytes_transferred;
      if (address && !address->FromSockAddr(storage.addr, storage.addr_len))
        result = ERR_FAILED;
    } else {
      result = MapSystemError(errno);
    }
    if (result != ERR_IO_PENDING)
      LogRead(result, buf->data(), storage.addr_len, storage.addr);
    return result;
  }
  
  int UDPSocketLibevent::InternalSendTo(IOBuffer* buf, int buf_len,
                                        const IPEndPoint* address) {
    SockaddrStorage storage;
    struct sockaddr* addr = storage.addr;
    if (!address) {
      addr = NULL;
      storage.addr_len = 0;
    } else {
      if (!address->ToSockAddr(storage.addr, &storage.addr_len)) {
        int result = ERR_FAILED;
        LogWrite(result, NULL, NULL);
        return result;
      }
    }
  
    int result = HANDLE_EINTR(sendto(socket_,
                              buf->data(),
                              buf_len,
                              0,
                              addr,
                              storage.addr_len));
    if (result < 0)
      result = MapSystemError(errno);
    if (result != ERR_IO_PENDING)
      LogWrite(result, buf->data(), address);
    return result;
  }
  
  int UDPSocketLibevent::SetSocketOptions() {
    int true_value = 1;
    if (socket_options_ & SOCKET_OPTION_REUSE_ADDRESS) {
      int rv = setsockopt(socket_, SOL_SOCKET, SO_REUSEADDR, &true_value,
                          sizeof(true_value));
      if (rv < 0)
        return MapSystemError(errno);
    }
    if (socket_options_ & SOCKET_OPTION_BROADCAST) {
      int rv;
  #if defined(OS_MACOSX)
      // SO_REUSEPORT on OSX permits multiple processes to each receive
      // UDP multicast or broadcast datagrams destined for the bound
      // port.
      rv = setsockopt(socket_, SOL_SOCKET, SO_REUSEPORT, &true_value,
                      sizeof(true_value));
  #else
      rv = setsockopt(socket_, SOL_SOCKET, SO_BROADCAST, &true_value,
                      sizeof(true_value));
  #endif  // defined(OS_MACOSX)
      if (rv < 0)
        return MapSystemError(errno);
    }
  
    if (!(socket_options_ & SOCKET_OPTION_MULTICAST_LOOP)) {
      int rv;
      if (addr_family_ == AF_INET) {
        u_char loop = 0;
        rv = setsockopt(socket_, IPPROTO_IP, IP_MULTICAST_LOOP,
                        &loop, sizeof(loop));
      } else {
        u_int loop = 0;
        rv = setsockopt(socket_, IPPROTO_IPV6, IPV6_MULTICAST_LOOP,
                        &loop, sizeof(loop));
      }
      if (rv < 0)
        return MapSystemError(errno);
    }
    if (multicast_time_to_live_ != IP_DEFAULT_MULTICAST_TTL) {
      int rv;
      if (addr_family_ == AF_INET) {
        u_char ttl = multicast_time_to_live_;
        rv = setsockopt(socket_, IPPROTO_IP, IP_MULTICAST_TTL,
                        &ttl, sizeof(ttl));
      } else {
        // Signed interger. -1 to use route default.
        int ttl = multicast_time_to_live_;
        rv = setsockopt(socket_, IPPROTO_IPV6, IPV6_MULTICAST_HOPS,
                        &ttl, sizeof(ttl));
      }
      if (rv < 0)
        return MapSystemError(errno);
    }
    return OK;
  }
  
  int UDPSocketLibevent::DoBind(const IPEndPoint& address) {
    SockaddrStorage storage;
    if (!address.ToSockAddr(storage.addr, &storage.addr_len))
      return ERR_ADDRESS_INVALID;
    int rv = bind(socket_, storage.addr, storage.addr_len);
    return rv < 0 ? MapSystemError(errno) : rv;
  }
  
  int UDPSocketLibevent::RandomBind(const IPEndPoint& address) {
    DCHECK(bind_type_ == DatagramSocket::RANDOM_BIND && !rand_int_cb_.is_null());
  
    // Construct IPAddressNumber of appropriate size (IPv4 or IPv6) of 0s.
    IPAddressNumber ip(address.address().size());
  
    for (int i = 0; i < kBindRetries; ++i) {
      int rv = DoBind(IPEndPoint(ip, rand_int_cb_.Run(kPortStart, kPortEnd)));
      if (rv == OK || rv != ERR_ADDRESS_IN_USE)
        return rv;
    }
    return DoBind(IPEndPoint(ip, 0));
  }
  
  int UDPSocketLibevent::JoinGroup(const IPAddressNumber& group_address) const {
    DCHECK(CalledOnValidThread());
    if (!is_connected())
      return ERR_SOCKET_NOT_CONNECTED;
  
    switch (group_address.size()) {
      case kIPv4AddressSize: {
        if (addr_family_ != AF_INET)
          return ERR_ADDRESS_INVALID;
        ip_mreq mreq;
        mreq.imr_interface.s_addr = INADDR_ANY;
        memcpy(&mreq.imr_multiaddr, &group_address[0], kIPv4AddressSize);
        int rv = setsockopt(socket_, IPPROTO_IP, IP_ADD_MEMBERSHIP,
                            &mreq, sizeof(mreq));
        if (rv < 0)
          return MapSystemError(errno);
        return OK;
      }
      case kIPv6AddressSize: {
        if (addr_family_ != AF_INET6)
          return ERR_ADDRESS_INVALID;
        ipv6_mreq mreq;
        mreq.ipv6mr_interface = 0;  // 0 indicates default multicast interface.
        memcpy(&mreq.ipv6mr_multiaddr, &group_address[0], kIPv6AddressSize);
        int rv = setsockopt(socket_, IPPROTO_IPV6, IPV6_JOIN_GROUP,
                            &mreq, sizeof(mreq));
        if (rv < 0)
          return MapSystemError(errno);
        return OK;
      }
      default:
        NOTREACHED() << "Invalid address family";
        return ERR_ADDRESS_INVALID;
    }
  }
  
  int UDPSocketLibevent::LeaveGroup(const IPAddressNumber& group_address) const {
    DCHECK(CalledOnValidThread());
  
    if (!is_connected())
      return ERR_SOCKET_NOT_CONNECTED;
  
    switch (group_address.size()) {
      case kIPv4AddressSize: {
        if (addr_family_ != AF_INET)
          return ERR_ADDRESS_INVALID;
        ip_mreq mreq;
        mreq.imr_interface.s_addr = INADDR_ANY;
        memcpy(&mreq.imr_multiaddr, &group_address[0], kIPv4AddressSize);
        int rv = setsockopt(socket_, IPPROTO_IP, IP_DROP_MEMBERSHIP,
                            &mreq, sizeof(mreq));
        if (rv < 0)
          return MapSystemError(errno);
        return OK;
      }
      case kIPv6AddressSize: {
        if (addr_family_ != AF_INET6)
          return ERR_ADDRESS_INVALID;
        ipv6_mreq mreq;
        mreq.ipv6mr_interface = 0;  // 0 indicates default multicast interface.
        memcpy(&mreq.ipv6mr_multiaddr, &group_address[0], kIPv6AddressSize);
        int rv = setsockopt(socket_, IPPROTO_IPV6, IPV6_LEAVE_GROUP,
                            &mreq, sizeof(mreq));
        if (rv < 0)
          return MapSystemError(errno);
        return OK;
      }
      default:
        NOTREACHED() << "Invalid address family";
        return ERR_ADDRESS_INVALID;
    }
  }
  
  int UDPSocketLibevent::SetMulticastTimeToLive(int time_to_live) {
    DCHECK(CalledOnValidThread());
    if (is_connected())
      return ERR_SOCKET_IS_CONNECTED;
  
    if (time_to_live < 0 || time_to_live > 255)
      return ERR_INVALID_ARGUMENT;
    multicast_time_to_live_ = time_to_live;
    return OK;
  }
  
  int UDPSocketLibevent::SetMulticastLoopbackMode(bool loopback) {
    DCHECK(CalledOnValidThread());
    if (is_connected())
      return ERR_SOCKET_IS_CONNECTED;
  
    if (loopback)
      socket_options_ |= SOCKET_OPTION_MULTICAST_LOOP;
    else
      socket_options_ &= ~SOCKET_OPTION_MULTICAST_LOOP;
    return OK;
  }
  }  // namespace net
  