net/udp/udp_socket_libevent.cc - chromium/src - Git at Google

 // Copyright (c) 2011 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/eintr_wrapper.h"
 #include "base/logging.h"
 #include "base/message_loop.h"
 #include "base/metrics/stats_counters.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_data_transfer_param.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),
           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),
           read_callback_(NULL),
           write_callback_(NULL),
           net_log_(BoundNetLog::Make(net_log, NetLog::SOURCE_UDP_SOCKET)) {
   scoped_refptr<NetLog::EventParameters> params;
   if (source.is_valid())
     params = new NetLogSourceParameter("source_dependency", source);
   net_log_.BeginEvent(NetLog::TYPE_SOCKET_ALIVE, params);
   if (bind_type == DatagramSocket::RANDOM_BIND)
     DCHECK(!rand_int_cb.is_null());
 }

 UDPSocketLibevent::~UDPSocketLibevent() {
   Close();
   net_log_.EndEvent(NetLog::TYPE_SOCKET_ALIVE, NULL);
 }

 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_ = NULL;
   recv_from_address_ = NULL;
   write_buf_ = NULL;
   write_buf_len_ = 0;
   write_callback_ = NULL;
   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()) {
     struct sockaddr_storage addr_storage;
     socklen_t addr_len = sizeof(addr_storage);
     struct sockaddr* addr = reinterpret_cast<struct sockaddr*>(&addr_storage);
     if (getpeername(socket_, addr, &addr_len))
       return MapSystemError(errno);
     scoped_ptr<IPEndPoint> address(new IPEndPoint());
     if (!address->FromSockAddr(addr, 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()) {
     struct sockaddr_storage addr_storage;
     socklen_t addr_len = sizeof(addr_storage);
     struct sockaddr* addr = reinterpret_cast<struct sockaddr*>(&addr_storage);
     if (getsockname(socket_, addr, &addr_len))
       return MapSystemError(errno);
     scoped_ptr<IPEndPoint> address(new IPEndPoint());
     if (!address->FromSockAddr(addr, addr_len))
       return ERR_FAILED;
     local_address_.reset(address.release());
   }

   *address = *local_address_;
   return OK;
 }

 int UDPSocketLibevent::Read(IOBuffer* buf,
                             int buf_len,
                             OldCompletionCallback* callback) {
   return RecvFrom(buf, buf_len, NULL, callback);
 }

 int UDPSocketLibevent::RecvFrom(IOBuffer* buf,
                                 int buf_len,
                                 IPEndPoint* address,
                                 OldCompletionCallback* callback) {
   DCHECK(CalledOnValidThread());
   DCHECK_NE(kInvalidSocket, socket_);
   DCHECK(!read_callback_);
   DCHECK(!recv_from_address_);
   DCHECK(callback);  // Synchronous operation not supported
   DCHECK_GT(buf_len, 0);

   int nread = InternalRecvFrom(buf, buf_len, address);
   if (nread != ERR_IO_PENDING)
     return nread;

   if (!MessageLoopForIO::current()->WatchFileDescriptor(
           socket_, true, 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,
                              OldCompletionCallback* callback) {
   return SendToOrWrite(buf, buf_len, NULL, callback);
 }

 int UDPSocketLibevent::SendTo(IOBuffer* buf,
                               int buf_len,
                               const IPEndPoint& address,
                               OldCompletionCallback* callback) {
   return SendToOrWrite(buf, buf_len, &address, callback);
 }

 int UDPSocketLibevent::SendToOrWrite(IOBuffer* buf,
                                      int buf_len,
                                      const IPEndPoint* address,
                                      OldCompletionCallback* callback) {
   DCHECK(CalledOnValidThread());
   DCHECK_NE(kInvalidSocket, socket_);
   DCHECK(!write_callback_);
   DCHECK(callback);  // Synchronous operation not supported
   DCHECK_GT(buf_len, 0);

   int result = InternalSendTo(buf, buf_len, address);
   if (result != ERR_IO_PENDING)
     return result;

   if (!MessageLoopForIO::current()->WatchFileDescriptor(
           socket_, true, 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,
       make_scoped_refptr(new NetLogStringParameter("address",
                                                    address.ToString())));
   int rv = InternalConnect(address);
   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;

   struct sockaddr_storage addr_storage;
   size_t addr_len = sizeof(addr_storage);
   struct sockaddr* addr = reinterpret_cast<struct sockaddr*>(&addr_storage);
   if (!address.ToSockAddr(addr, &addr_len))
     return ERR_FAILED;

   rv = HANDLE_EINTR(connect(socket_, addr, 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 = 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::DoReadCallback(int rv) {
   DCHECK_NE(rv, ERR_IO_PENDING);
   DCHECK(read_callback_);

   // since Run may result in Read being called, clear read_callback_ up front.
   OldCompletionCallback* c = read_callback_;
   read_callback_ = NULL;
   c->Run(rv);
 }

 void UDPSocketLibevent::DoWriteCallback(int rv) {
   DCHECK_NE(rv, ERR_IO_PENDING);
   DCHECK(write_callback_);

   // since Run may result in Write being called, clear write_callback_ up front.
   OldCompletionCallback* c = write_callback_;
   write_callback_ = NULL;
   c->Run(rv);
 }

 void UDPSocketLibevent::DidCompleteRead() {
   int result = InternalRecvFrom(read_buf_, 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,
         make_scoped_refptr(
             new UDPDataTransferNetLogParam(
                 result, bytes, net_log_.IsLoggingBytes(),
                 is_address_valid ? &address : NULL)));
   }

   base::StatsCounter read_bytes("udp.read_bytes");
   read_bytes.Add(result);
 }

 int UDPSocketLibevent::CreateSocket(const IPEndPoint& address) {
   socket_ = socket(address.GetFamily(), 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_, 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,
         make_scoped_refptr(
             new UDPDataTransferNetLogParam(result, bytes,
                                            net_log_.IsLoggingBytes(),
                                            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;

   struct sockaddr_storage addr_storage;
   socklen_t addr_len = sizeof(addr_storage);
   struct sockaddr* addr = reinterpret_cast<struct sockaddr*>(&addr_storage);

   bytes_transferred =
       HANDLE_EINTR(recvfrom(socket_,
                             buf->data(),
                             buf_len,
                             flags,
                             addr,
                             &addr_len));
   int result;
   if (bytes_transferred >= 0) {
     result = bytes_transferred;
     if (address && !address->FromSockAddr(addr, addr_len))
       result = ERR_FAILED;
   } else {
     result = MapSystemError(errno);
   }
   if (result != ERR_IO_PENDING)
     LogRead(result, buf->data(), addr_len, addr);
   return result;
 }

 int UDPSocketLibevent::InternalSendTo(IOBuffer* buf, int buf_len,
                                       const IPEndPoint* address) {
   struct sockaddr_storage addr_storage;
   size_t addr_len = sizeof(addr_storage);
   struct sockaddr* addr = reinterpret_cast<struct sockaddr*>(&addr_storage);

   if (!address) {
     addr = NULL;
     addr_len = 0;
   } else {
     if (!address->ToSockAddr(addr, &addr_len)) {
       int result = ERR_FAILED;
       LogWrite(result, NULL, NULL);
       return result;
     }
   }

   int result = HANDLE_EINTR(sendto(socket_,
                             buf->data(),
                             buf_len,
                             0,
                             addr,
                             addr_len));
   if (result < 0)
     result = MapSystemError(errno);
   if (result != ERR_IO_PENDING)
     LogWrite(result, buf->data(), address);
   return result;
 }

 int UDPSocketLibevent::DoBind(const IPEndPoint& address) {
   struct sockaddr_storage addr_storage;
   size_t addr_len = sizeof(addr_storage);
   struct sockaddr* addr = reinterpret_cast<struct sockaddr*>(&addr_storage);
   if (!address.ToSockAddr(addr, &addr_len))
     return ERR_UNEXPECTED;
   int rv = bind(socket_, addr, 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));
 }

 }  // namespace net
	// Copyright (c) 2011 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/eintr_wrapper.h"
	#include "base/logging.h"
	#include "base/message_loop.h"
	#include "base/metrics/stats_counters.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_data_transfer_param.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),
	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),
	read_callback_(NULL),
	write_callback_(NULL),
	net_log_(BoundNetLog::Make(net_log, NetLog::SOURCE_UDP_SOCKET)) {
	scoped_refptr<NetLog::EventParameters> params;
	if (source.is_valid())
	params = new NetLogSourceParameter("source_dependency", source);
	net_log_.BeginEvent(NetLog::TYPE_SOCKET_ALIVE, params);
	if (bind_type == DatagramSocket::RANDOM_BIND)
	DCHECK(!rand_int_cb.is_null());
	}

	UDPSocketLibevent::~UDPSocketLibevent() {
	Close();
	net_log_.EndEvent(NetLog::TYPE_SOCKET_ALIVE, NULL);
	}

	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_ = NULL;
	recv_from_address_ = NULL;
	write_buf_ = NULL;
	write_buf_len_ = 0;
	write_callback_ = NULL;
	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()) {
	struct sockaddr_storage addr_storage;
	socklen_t addr_len = sizeof(addr_storage);
	struct sockaddr* addr = reinterpret_cast<struct sockaddr*>(&addr_storage);
	if (getpeername(socket_, addr, &addr_len))
	return MapSystemError(errno);
	scoped_ptr<IPEndPoint> address(new IPEndPoint());
	if (!address->FromSockAddr(addr, 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()) {
	struct sockaddr_storage addr_storage;
	socklen_t addr_len = sizeof(addr_storage);
	struct sockaddr* addr = reinterpret_cast<struct sockaddr*>(&addr_storage);
	if (getsockname(socket_, addr, &addr_len))
	return MapSystemError(errno);
	scoped_ptr<IPEndPoint> address(new IPEndPoint());
	if (!address->FromSockAddr(addr, addr_len))
	return ERR_FAILED;
	local_address_.reset(address.release());
	}

	address = local_address_;
	return OK;
	}

	int UDPSocketLibevent::Read(IOBuffer* buf,
	int buf_len,
	OldCompletionCallback* callback) {
	return RecvFrom(buf, buf_len, NULL, callback);
	}

	int UDPSocketLibevent::RecvFrom(IOBuffer* buf,
	int buf_len,
	IPEndPoint* address,
	OldCompletionCallback* callback) {
	DCHECK(CalledOnValidThread());
	DCHECK_NE(kInvalidSocket, socket_);
	DCHECK(!read_callback_);
	DCHECK(!recv_from_address_);
	DCHECK(callback); // Synchronous operation not supported
	DCHECK_GT(buf_len, 0);

	int nread = InternalRecvFrom(buf, buf_len, address);
	if (nread != ERR_IO_PENDING)
	return nread;

	if (!MessageLoopForIO::current()->WatchFileDescriptor(
	socket_, true, 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,
	OldCompletionCallback* callback) {
	return SendToOrWrite(buf, buf_len, NULL, callback);
	}

	int UDPSocketLibevent::SendTo(IOBuffer* buf,
	int buf_len,
	const IPEndPoint& address,
	OldCompletionCallback* callback) {
	return SendToOrWrite(buf, buf_len, &address, callback);
	}

	int UDPSocketLibevent::SendToOrWrite(IOBuffer* buf,
	int buf_len,
	const IPEndPoint* address,
	OldCompletionCallback* callback) {
	DCHECK(CalledOnValidThread());
	DCHECK_NE(kInvalidSocket, socket_);
	DCHECK(!write_callback_);
	DCHECK(callback); // Synchronous operation not supported
	DCHECK_GT(buf_len, 0);

	int result = InternalSendTo(buf, buf_len, address);
	if (result != ERR_IO_PENDING)
	return result;

	if (!MessageLoopForIO::current()->WatchFileDescriptor(
	socket_, true, 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,
	make_scoped_refptr(new NetLogStringParameter("address",
	address.ToString())));
	int rv = InternalConnect(address);
	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;

	struct sockaddr_storage addr_storage;
	size_t addr_len = sizeof(addr_storage);
	struct sockaddr* addr = reinterpret_cast<struct sockaddr*>(&addr_storage);
	if (!address.ToSockAddr(addr, &addr_len))
	return ERR_FAILED;

	rv = HANDLE_EINTR(connect(socket_, addr, 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 = 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::DoReadCallback(int rv) {
	DCHECK_NE(rv, ERR_IO_PENDING);
	DCHECK(read_callback_);

	// since Run may result in Read being called, clear read_callback_ up front.
	OldCompletionCallback* c = read_callback_;
	read_callback_ = NULL;
	c->Run(rv);
	}

	void UDPSocketLibevent::DoWriteCallback(int rv) {
	DCHECK_NE(rv, ERR_IO_PENDING);
	DCHECK(write_callback_);

	// since Run may result in Write being called, clear write_callback_ up front.
	OldCompletionCallback* c = write_callback_;
	write_callback_ = NULL;
	c->Run(rv);
	}

	void UDPSocketLibevent::DidCompleteRead() {
	int result = InternalRecvFrom(read_buf_, 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,
	make_scoped_refptr(
	new UDPDataTransferNetLogParam(
	result, bytes, net_log_.IsLoggingBytes(),
	is_address_valid ? &address : NULL)));
	}

	base::StatsCounter read_bytes("udp.read_bytes");
	read_bytes.Add(result);
	}

	int UDPSocketLibevent::CreateSocket(const IPEndPoint& address) {
	socket_ = socket(address.GetFamily(), 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_, 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,
	make_scoped_refptr(
	new UDPDataTransferNetLogParam(result, bytes,
	net_log_.IsLoggingBytes(),
	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;

	struct sockaddr_storage addr_storage;
	socklen_t addr_len = sizeof(addr_storage);
	struct sockaddr* addr = reinterpret_cast<struct sockaddr*>(&addr_storage);

	bytes_transferred =
	HANDLE_EINTR(recvfrom(socket_,
	buf->data(),
	buf_len,
	flags,
	addr,
	&addr_len));
	int result;
	if (bytes_transferred >= 0) {
	result = bytes_transferred;
	if (address && !address->FromSockAddr(addr, addr_len))
	result = ERR_FAILED;
	} else {
	result = MapSystemError(errno);
	}
	if (result != ERR_IO_PENDING)
	LogRead(result, buf->data(), addr_len, addr);
	return result;
	}

	int UDPSocketLibevent::InternalSendTo(IOBuffer* buf, int buf_len,
	const IPEndPoint* address) {
	struct sockaddr_storage addr_storage;
	size_t addr_len = sizeof(addr_storage);
	struct sockaddr* addr = reinterpret_cast<struct sockaddr*>(&addr_storage);

	if (!address) {
	addr = NULL;
	addr_len = 0;
	} else {
	if (!address->ToSockAddr(addr, &addr_len)) {
	int result = ERR_FAILED;
	LogWrite(result, NULL, NULL);
	return result;
	}
	}

	int result = HANDLE_EINTR(sendto(socket_,
	buf->data(),
	buf_len,
	0,
	addr,
	addr_len));
	if (result < 0)
	result = MapSystemError(errno);
	if (result != ERR_IO_PENDING)
	LogWrite(result, buf->data(), address);
	return result;
	}

	int UDPSocketLibevent::DoBind(const IPEndPoint& address) {
	struct sockaddr_storage addr_storage;
	size_t addr_len = sizeof(addr_storage);
	struct sockaddr* addr = reinterpret_cast<struct sockaddr*>(&addr_storage);
	if (!address.ToSockAddr(addr, &addr_len))
	return ERR_UNEXPECTED;
	int rv = bind(socket_, addr, 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));
	}

	} // namespace net