Refactor tcp socket.

This is a pre-requisite for http server refactoring,
https://codereview.chromium.org/296053012/.

1) Introduce SocketLibevent for tcp and unix domain sockets.
2) TCPSocketLibevent utilizes SocketLibevent.
3) For backward compatibility, TCPSocketLibevent::AdoptConnectedSocket()
   allows the empty address.

BUG=371906

Review URL: https://codereview.chromium.org/348803003

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@282121 0039d316-1c4b-4281-b951-d872f2087c98
parent 273ae5ab
......@@ -7,6 +7,7 @@
#include <queue>
#include <string>
#include "base/location.h"
#include "base/logging.h"
#include "base/memory/linked_ptr.h"
#include "base/memory/ref_counted.h"
......
......@@ -196,14 +196,10 @@ int TCPSocket::Listen(const std::string& address,
DCHECK(!socket_.get());
socket_mode_ = SERVER;
scoped_ptr<net::IPEndPoint> bind_address(new net::IPEndPoint());
if (!StringAndPortToIPEndPoint(address, port, bind_address.get()))
return net::ERR_INVALID_ARGUMENT;
if (!server_socket_.get()) {
server_socket_.reset(new net::TCPServerSocket(NULL, net::NetLog::Source()));
}
int result = server_socket_->Listen(*bind_address, backlog);
int result = server_socket_->ListenWithAddressAndPort(address, port, backlog);
if (result)
*error_msg = kSocketListenError;
return result;
......
......@@ -435,6 +435,8 @@ component("net") {
if (is_win) {
sources -= [
"http/http_auth_handler_ntlm_portable.cc",
"socket/socket_libevent.cc",
"socket/socket_libevent.h",
"socket/tcp_socket_libevent.cc",
"socket/tcp_socket_libevent.h",
"udp/udp_socket_libevent.cc",
......
......@@ -5,6 +5,7 @@
#include "net/base/net_util.h"
#include <errno.h>
#include <string.h>
#include <algorithm>
#include <iterator>
......@@ -501,6 +502,18 @@ bool IsIPAddressReserved(const IPAddressNumber& host_addr) {
return false;
}
SockaddrStorage::SockaddrStorage(const SockaddrStorage& other)
: addr_len(other.addr_len),
addr(reinterpret_cast<struct sockaddr*>(&addr_storage)) {
memcpy(addr, other.addr, addr_len);
}
void SockaddrStorage::operator=(const SockaddrStorage& other) {
addr_len = other.addr_len;
// addr is already set to &this->addr_storage by default ctor.
memcpy(addr, other.addr, addr_len);
}
// Extracts the address and port portions of a sockaddr.
bool GetIPAddressFromSockAddr(const struct sockaddr* sock_addr,
socklen_t sock_addr_len,
......
......@@ -114,6 +114,9 @@ NET_EXPORT bool IsIPAddressReserved(const IPAddressNumber& address);
struct SockaddrStorage {
SockaddrStorage() : addr_len(sizeof(addr_storage)),
addr(reinterpret_cast<struct sockaddr*>(&addr_storage)) {}
SockaddrStorage(const SockaddrStorage& other);
void operator=(const SockaddrStorage& other);
struct sockaddr_storage addr_storage;
socklen_t addr_len;
struct sockaddr* const addr;
......
......@@ -392,6 +392,8 @@
[ 'OS == "win"', {
'sources!': [
'http/http_auth_handler_ntlm_portable.cc',
'socket/socket_libevent.cc',
'socket/socket_libevent.h',
'socket/tcp_socket_libevent.cc',
'socket/tcp_socket_libevent.h',
'udp/udp_socket_libevent.cc',
......
......@@ -947,9 +947,12 @@
'socket/client_socket_pool_manager_impl.h',
'socket/nss_ssl_util.cc',
'socket/nss_ssl_util.h',
'socket/server_socket.cc',
'socket/server_socket.h',
'socket/socket_descriptor.cc',
'socket/socket_descriptor.h',
'socket/socket_libevent.cc',
'socket/socket_libevent.h',
'socket/socket_net_log_params.cc',
'socket/socket_net_log_params.h',
'socket/socks5_client_socket.cc',
......
// Copyright 2014 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/socket/server_socket.h"
#include "net/base/ip_endpoint.h"
#include "net/base/net_errors.h"
#include "net/base/net_util.h"
namespace net {
ServerSocket::ServerSocket() {
}
ServerSocket::~ServerSocket() {
}
int ServerSocket::ListenWithAddressAndPort(const std::string& address_string,
int port,
int backlog) {
IPAddressNumber address_number;
if (!ParseIPLiteralToNumber(address_string, &address_number)) {
return ERR_ADDRESS_INVALID;
}
return Listen(IPEndPoint(address_number, port), backlog);
}
} // namespace net
......@@ -5,6 +5,8 @@
#ifndef NET_SOCKET_SERVER_SOCKET_H_
#define NET_SOCKET_SERVER_SOCKET_H_
#include <string>
#include "base/memory/scoped_ptr.h"
#include "net/base/completion_callback.h"
#include "net/base/net_export.h"
......@@ -16,17 +18,25 @@ class StreamSocket;
class NET_EXPORT ServerSocket {
public:
ServerSocket() { }
virtual ~ServerSocket() { }
ServerSocket();
virtual ~ServerSocket();
// Bind the socket and start listening. Destroy the socket to stop
// Binds the socket and starts listening. Destroy the socket to stop
// listening.
virtual int Listen(const net::IPEndPoint& address, int backlog) = 0;
// Binds the socket with address and port, and starts listening. It expects
// a valid IPv4 or IPv6 address. Otherwise, it returns ERR_ADDRESS_INVALID.
// Subclasses may override this function if |address_string| is in a different
// format, for example, unix domain socket path.
virtual int ListenWithAddressAndPort(const std::string& address_string,
int port,
int backlog);
// Gets current address the socket is bound to.
virtual int GetLocalAddress(IPEndPoint* address) const = 0;
// Accept connection. Callback is called when new connection is
// Accepts connection. Callback is called when new connection is
// accepted.
virtual int Accept(scoped_ptr<StreamSocket>* socket,
const CompletionCallback& callback) = 0;
......
// Copyright 2014 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/socket/socket_libevent.h"
#include <errno.h>
#include <netinet/in.h>
#include <sys/socket.h>
#include "base/callback_helpers.h"
#include "base/logging.h"
#include "base/posix/eintr_wrapper.h"
#include "net/base/io_buffer.h"
#include "net/base/ip_endpoint.h"
#include "net/base/net_errors.h"
#include "net/base/net_util.h"
namespace net {
namespace {
int MapAcceptError(int os_error) {
switch (os_error) {
// If the client aborts the connection before the server calls accept,
// POSIX specifies accept should fail with ECONNABORTED. The server can
// ignore the error and just call accept again, so we map the error to
// ERR_IO_PENDING. See UNIX Network Programming, Vol. 1, 3rd Ed., Sec.
// 5.11, "Connection Abort before accept Returns".
case ECONNABORTED:
return ERR_IO_PENDING;
default:
return MapSystemError(os_error);
}
}
int MapConnectError(int os_error) {
switch (os_error) {
case EINPROGRESS:
return ERR_IO_PENDING;
case EACCES:
return ERR_NETWORK_ACCESS_DENIED;
case ETIMEDOUT:
return ERR_CONNECTION_TIMED_OUT;
default: {
int net_error = MapSystemError(os_error);
if (net_error == ERR_FAILED)
return ERR_CONNECTION_FAILED; // More specific than ERR_FAILED.
return net_error;
}
}
}
} // namespace
SocketLibevent::SocketLibevent()
: socket_fd_(kInvalidSocket),
read_buf_len_(0),
write_buf_len_(0),
waiting_connect_(false) {
}
SocketLibevent::~SocketLibevent() {
Close();
}
int SocketLibevent::Open(int address_family) {
DCHECK(thread_checker_.CalledOnValidThread());
DCHECK_EQ(kInvalidSocket, socket_fd_);
DCHECK(address_family == AF_INET ||
address_family == AF_INET6 ||
address_family == AF_UNIX);
socket_fd_ = CreatePlatformSocket(
address_family,
SOCK_STREAM,
address_family == AF_UNIX ? 0 : IPPROTO_TCP);
if (socket_fd_ < 0) {
PLOG(ERROR) << "CreatePlatformSocket() returned an error, errno=" << errno;
return MapSystemError(errno);
}
if (SetNonBlocking(socket_fd_)) {
int rv = MapSystemError(errno);
Close();
return rv;
}
return OK;
}
int SocketLibevent::AdoptConnectedSocket(SocketDescriptor socket,
const SockaddrStorage& address) {
DCHECK(thread_checker_.CalledOnValidThread());
DCHECK_EQ(kInvalidSocket, socket_fd_);
socket_fd_ = socket;
if (SetNonBlocking(socket_fd_)) {
int rv = MapSystemError(errno);
Close();
return rv;
}
SetPeerAddress(address);
return OK;
}
int SocketLibevent::Bind(const SockaddrStorage& address) {
DCHECK(thread_checker_.CalledOnValidThread());
DCHECK_NE(kInvalidSocket, socket_fd_);
int rv = bind(socket_fd_, address.addr, address.addr_len);
if (rv < 0) {
PLOG(ERROR) << "bind() returned an error, errno=" << errno;
return MapSystemError(errno);
}
return OK;
}
int SocketLibevent::Listen(int backlog) {
DCHECK(thread_checker_.CalledOnValidThread());
DCHECK_NE(kInvalidSocket, socket_fd_);
DCHECK_LT(0, backlog);
int rv = listen(socket_fd_, backlog);
if (rv < 0) {
PLOG(ERROR) << "listen() returned an error, errno=" << errno;
return MapSystemError(errno);
}
return OK;
}
int SocketLibevent::Accept(scoped_ptr<SocketLibevent>* socket,
const CompletionCallback& callback) {
DCHECK(thread_checker_.CalledOnValidThread());
DCHECK_NE(kInvalidSocket, socket_fd_);
DCHECK(accept_callback_.is_null());
DCHECK(socket);
DCHECK(!callback.is_null());
int rv = DoAccept(socket);
if (rv != ERR_IO_PENDING)
return rv;
if (!base::MessageLoopForIO::current()->WatchFileDescriptor(
socket_fd_, true, base::MessageLoopForIO::WATCH_READ,
&accept_socket_watcher_, this)) {
PLOG(ERROR) << "WatchFileDescriptor failed on accept, errno " << errno;
return MapSystemError(errno);
}
accept_socket_ = socket;
accept_callback_ = callback;
return ERR_IO_PENDING;
}
int SocketLibevent::Connect(const SockaddrStorage& address,
const CompletionCallback& callback) {
DCHECK(thread_checker_.CalledOnValidThread());
DCHECK_NE(kInvalidSocket, socket_fd_);
DCHECK(!waiting_connect_);
DCHECK(!callback.is_null());
SetPeerAddress(address);
int rv = DoConnect();
if (rv != ERR_IO_PENDING)
return rv;
if (!base::MessageLoopForIO::current()->WatchFileDescriptor(
socket_fd_, true, base::MessageLoopForIO::WATCH_WRITE,
&write_socket_watcher_, this)) {
PLOG(ERROR) << "WatchFileDescriptor failed on connect, errno " << errno;
return MapSystemError(errno);
}
write_callback_ = callback;
waiting_connect_ = true;
return ERR_IO_PENDING;
}
bool SocketLibevent::IsConnected() const {
DCHECK(thread_checker_.CalledOnValidThread());
if (socket_fd_ == kInvalidSocket || waiting_connect_)
return false;
// Checks if connection is alive.
char c;
int rv = HANDLE_EINTR(recv(socket_fd_, &c, 1, MSG_PEEK));
if (rv == 0)
return false;
if (rv == -1 && errno != EAGAIN && errno != EWOULDBLOCK)
return false;
return true;
}
bool SocketLibevent::IsConnectedAndIdle() const {
DCHECK(thread_checker_.CalledOnValidThread());
if (socket_fd_ == kInvalidSocket || waiting_connect_)
return false;
// Check if connection is alive and we haven't received any data
// unexpectedly.
char c;
int rv = HANDLE_EINTR(recv(socket_fd_, &c, 1, MSG_PEEK));
if (rv >= 0)
return false;
if (errno != EAGAIN && errno != EWOULDBLOCK)
return false;
return true;
}
int SocketLibevent::Read(IOBuffer* buf,
int buf_len,
const CompletionCallback& callback) {
DCHECK(thread_checker_.CalledOnValidThread());
DCHECK_NE(kInvalidSocket, socket_fd_);
DCHECK(!waiting_connect_);
DCHECK(read_callback_.is_null());
// Synchronous operation not supported
DCHECK(!callback.is_null());
DCHECK_LT(0, buf_len);
int rv = DoRead(buf, buf_len);
if (rv != ERR_IO_PENDING)
return rv;
if (!base::MessageLoopForIO::current()->WatchFileDescriptor(
socket_fd_, true, base::MessageLoopForIO::WATCH_READ,
&read_socket_watcher_, this)) {
PLOG(ERROR) << "WatchFileDescriptor failed on read, errno " << errno;
return MapSystemError(errno);
}
read_buf_ = buf;
read_buf_len_ = buf_len;
read_callback_ = callback;
return ERR_IO_PENDING;
}
int SocketLibevent::Write(IOBuffer* buf,
int buf_len,
const CompletionCallback& callback) {
DCHECK(thread_checker_.CalledOnValidThread());
DCHECK_NE(kInvalidSocket, socket_fd_);
DCHECK(!waiting_connect_);
DCHECK(write_callback_.is_null());
// Synchronous operation not supported
DCHECK(!callback.is_null());
DCHECK_LT(0, buf_len);
int rv = DoWrite(buf, buf_len);
if (rv == ERR_IO_PENDING)
rv = WaitForWrite(buf, buf_len, callback);
return rv;
}
int SocketLibevent::WaitForWrite(IOBuffer* buf,
int buf_len,
const CompletionCallback& callback) {
DCHECK(thread_checker_.CalledOnValidThread());
DCHECK_NE(kInvalidSocket, socket_fd_);
DCHECK(write_callback_.is_null());
// Synchronous operation not supported
DCHECK(!callback.is_null());
DCHECK_LT(0, buf_len);
if (!base::MessageLoopForIO::current()->WatchFileDescriptor(
socket_fd_, true, base::MessageLoopForIO::WATCH_WRITE,
&write_socket_watcher_, this)) {
PLOG(ERROR) << "WatchFileDescriptor failed on write, errno " << errno;
return MapSystemError(errno);
}
write_buf_ = buf;
write_buf_len_ = buf_len;
write_callback_ = callback;
return ERR_IO_PENDING;
}
int SocketLibevent::GetLocalAddress(SockaddrStorage* address) const {
DCHECK(thread_checker_.CalledOnValidThread());
DCHECK(address);
if (getsockname(socket_fd_, address->addr, &address->addr_len) < 0)
return MapSystemError(errno);
return OK;
}
int SocketLibevent::GetPeerAddress(SockaddrStorage* address) const {
DCHECK(thread_checker_.CalledOnValidThread());
DCHECK(address);
if (!HasPeerAddress())
return ERR_SOCKET_NOT_CONNECTED;
*address = *peer_address_;
return OK;
}
void SocketLibevent::SetPeerAddress(const SockaddrStorage& address) {
DCHECK(thread_checker_.CalledOnValidThread());
// |peer_address_| will be non-NULL if Connect() has been called. Unless
// Close() is called to reset the internal state, a second call to Connect()
// is not allowed.
// Please note that we don't allow a second Connect() even if the previous
// Connect() has failed. Connecting the same |socket_| again after a
// connection attempt failed results in unspecified behavior according to
// POSIX.
DCHECK(!peer_address_);
peer_address_.reset(new SockaddrStorage(address));
}
bool SocketLibevent::HasPeerAddress() const {
DCHECK(thread_checker_.CalledOnValidThread());
return peer_address_ != NULL;
}
void SocketLibevent::Close() {
DCHECK(thread_checker_.CalledOnValidThread());
bool ok = accept_socket_watcher_.StopWatchingFileDescriptor();
DCHECK(ok);
ok = read_socket_watcher_.StopWatchingFileDescriptor();
DCHECK(ok);
ok = write_socket_watcher_.StopWatchingFileDescriptor();
DCHECK(ok);
if (socket_fd_ != kInvalidSocket) {
if (IGNORE_EINTR(close(socket_fd_)) < 0)
PLOG(ERROR) << "close() returned an error, errno=" << errno;
socket_fd_ = kInvalidSocket;
}
if (!accept_callback_.is_null()) {
accept_socket_ = NULL;
accept_callback_.Reset();
}
if (!read_callback_.is_null()) {
read_buf_ = NULL;
read_buf_len_ = 0;
read_callback_.Reset();
}
if (!write_callback_.is_null()) {
write_buf_ = NULL;
write_buf_len_ = 0;
write_callback_.Reset();
}
waiting_connect_ = false;
peer_address_.reset();
}
void SocketLibevent::OnFileCanReadWithoutBlocking(int fd) {
DCHECK(!accept_callback_.is_null() || !read_callback_.is_null());
if (!accept_callback_.is_null()) {
AcceptCompleted();
} else { // !read_callback_.is_null()
ReadCompleted();
}
}
void SocketLibevent::OnFileCanWriteWithoutBlocking(int fd) {
DCHECK(!write_callback_.is_null());
if (waiting_connect_) {
ConnectCompleted();
} else {
WriteCompleted();
}
}
int SocketLibevent::DoAccept(scoped_ptr<SocketLibevent>* socket) {
SockaddrStorage new_peer_address;
int new_socket = HANDLE_EINTR(accept(socket_fd_,
new_peer_address.addr,
&new_peer_address.addr_len));
if (new_socket < 0)
return MapAcceptError(errno);
scoped_ptr<SocketLibevent> accepted_socket(new SocketLibevent);
int rv = accepted_socket->AdoptConnectedSocket(new_socket, new_peer_address);
if (rv != OK)
return rv;
*socket = accepted_socket.Pass();
return OK;
}
void SocketLibevent::AcceptCompleted() {
DCHECK(accept_socket_);
int rv = DoAccept(accept_socket_);
if (rv == ERR_IO_PENDING)
return;
bool ok = accept_socket_watcher_.StopWatchingFileDescriptor();
DCHECK(ok);
accept_socket_ = NULL;
base::ResetAndReturn(&accept_callback_).Run(rv);
}
int SocketLibevent::DoConnect() {
int rv = HANDLE_EINTR(connect(socket_fd_,
peer_address_->addr,
peer_address_->addr_len));
DCHECK_GE(0, rv);
return rv == 0 ? OK : MapConnectError(errno);
}
void SocketLibevent::ConnectCompleted() {
// Get the error that connect() completed with.
int os_error = 0;
socklen_t len = sizeof(os_error);
if (getsockopt(socket_fd_, SOL_SOCKET, SO_ERROR, &os_error, &len) == 0) {
// TCPSocketLibevent expects errno to be set.
errno = os_error;
}
int rv = MapConnectError(errno);
if (rv == ERR_IO_PENDING)
return;
bool ok = write_socket_watcher_.StopWatchingFileDescriptor();
DCHECK(ok);
waiting_connect_ = false;
base::ResetAndReturn(&write_callback_).Run(rv);
}
int SocketLibevent::DoRead(IOBuffer* buf, int buf_len) {
int rv = HANDLE_EINTR(read(socket_fd_, buf->data(), buf_len));
return rv >= 0 ? rv : MapSystemError(errno);
}
void SocketLibevent::ReadCompleted() {
int rv = DoRead(read_buf_, read_buf_len_);
if (rv == ERR_IO_PENDING)
return;
bool ok = read_socket_watcher_.StopWatchingFileDescriptor();
DCHECK(ok);
read_buf_ = NULL;
read_buf_len_ = 0;
base::ResetAndReturn(&read_callback_).Run(rv);
}
int SocketLibevent::DoWrite(IOBuffer* buf, int buf_len) {
int rv = HANDLE_EINTR(write(socket_fd_, buf->data(), buf_len));
return rv >= 0 ? rv : MapSystemError(errno);
}
void SocketLibevent::WriteCompleted() {
int rv = DoWrite(write_buf_, write_buf_len_);
if (rv == ERR_IO_PENDING)
return;
bool ok = write_socket_watcher_.StopWatchingFileDescriptor();
DCHECK(ok);
write_buf_ = NULL;
write_buf_len_ = 0;
base::ResetAndReturn(&write_callback_).Run(rv);
}
} // namespace net
// Copyright 2014 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.
#ifndef NET_SOCKET_SOCKET_LIBEVENT_H_
#define NET_SOCKET_SOCKET_LIBEVENT_H_
#include "base/basictypes.h"
#include "base/compiler_specific.h"
#include "base/macros.h"
#include "base/memory/ref_counted.h"
#include "base/memory/scoped_ptr.h"
#include "base/message_loop/message_loop.h"
#include "base/threading/thread_checker.h"
#include "net/base/completion_callback.h"
#include "net/base/net_util.h"
#include "net/socket/socket_descriptor.h"
namespace net {
class IOBuffer;
class IPEndPoint;
// Socket class to provide asynchronous read/write operations on top of the
// posix socket api. It supports AF_INET, AF_INET6, and AF_UNIX addresses.
class SocketLibevent : public base::MessageLoopForIO::Watcher {
public:
SocketLibevent();
virtual ~SocketLibevent();
// Opens a socket and returns net::OK if |address_family| is AF_INET, AF_INET6
// or AF_UNIX. Otherwise, it does DCHECK() and returns a net error.
int Open(int address_family);
// Takes ownership of |socket|.
int AdoptConnectedSocket(SocketDescriptor socket,
const SockaddrStorage& peer_address);
int Bind(const SockaddrStorage& address);
int Listen(int backlog);
int Accept(scoped_ptr<SocketLibevent>* socket,
const CompletionCallback& callback);
// Connects socket. On non-ERR_IO_PENDING error, sets errno and returns a net
// error code. On ERR_IO_PENDING, |callback| is called with a net error code,
// not errno, though errno is set if connect event happens with error.
// TODO(byungchul): Need more robust way to pass system errno.
int Connect(const SockaddrStorage& address,
const CompletionCallback& callback);
bool IsConnected() const;
bool IsConnectedAndIdle() const;
// Multiple outstanding requests of the same type are not supported.
// Full duplex mode (reading and writing at the same time) is supported.
// On error which is not ERR_IO_PENDING, sets errno and returns a net error
// code. On ERR_IO_PENDING, |callback| is called with a net error code, not
// errno, though errno is set if read or write events happen with error.
// TODO(byungchul): Need more robust way to pass system errno.
int Read(IOBuffer* buf, int buf_len, const CompletionCallback& callback);
int Write(IOBuffer* buf, int buf_len, const CompletionCallback& callback);
// Waits for next write event. This is called by TCPsocketLibevent for TCP
// fastopen after sending first data. Returns ERR_IO_PENDING if it starts
// waiting for write event successfully. Otherwise, returns a net error code.
// It must not be called after Write() because Write() calls it internally.
int WaitForWrite(IOBuffer* buf, int buf_len,
const CompletionCallback& callback);
int GetLocalAddress(SockaddrStorage* address) const;
int GetPeerAddress(SockaddrStorage* address) const;
void SetPeerAddress(const SockaddrStorage& address);
// Returns true if peer address has been set regardless of socket state.
bool HasPeerAddress() const;
void Close();
SocketDescriptor socket_fd() const { return socket_fd_; }
private:
// base::MessageLoopForIO::Watcher methods.
virtual void OnFileCanReadWithoutBlocking(int fd) OVERRIDE;
virtual void OnFileCanWriteWithoutBlocking(int fd) OVERRIDE;
int DoAccept(scoped_ptr<SocketLibevent>* socket);
void AcceptCompleted();
int DoConnect();
void ConnectCompleted();
int DoRead(IOBuffer* buf, int buf_len);
void ReadCompleted();
int DoWrite(IOBuffer* buf, int buf_len);
void WriteCompleted();
SocketDescriptor socket_fd_;
base::MessageLoopForIO::FileDescriptorWatcher accept_socket_watcher_;
scoped_ptr<SocketLibevent>* accept_socket_;
CompletionCallback accept_callback_;
base::MessageLoopForIO::FileDescriptorWatcher read_socket_watcher_;
scoped_refptr<IOBuffer> read_buf_;
int read_buf_len_;
// External callback; called when read is complete.
CompletionCallback read_callback_;
base::MessageLoopForIO::FileDescriptorWatcher write_socket_watcher_;
scoped_refptr<IOBuffer> write_buf_;
int write_buf_len_;
// External callback; called when write or connect is complete.
CompletionCallback write_callback_;
// A connect operation is pending. In this case, |write_callback_| needs to be
// called when connect is complete.
bool waiting_connect_;
scoped_ptr<SockaddrStorage> peer_address_;
base::ThreadChecker thread_checker_;
DISALLOW_COPY_AND_ASSIGN(SocketLibevent);
};
} // namespace net
#endif // NET_SOCKET_SOCKET_LIBEVENT_H_
......@@ -6,6 +6,7 @@
#include "base/file_util.h"
#include "base/files/file_path.h"
#include "base/location.h"
#include "base/memory/ref_counted.h"
#include "base/threading/worker_pool.h"
......
......@@ -5,18 +5,14 @@
#include "net/socket/tcp_socket.h"
#include <errno.h>
#include <fcntl.h>
#include <netdb.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <sys/socket.h>
#include "base/callback_helpers.h"
#include "base/bind.h"
#include "base/logging.h"
#include "base/metrics/histogram.h"
#include "base/metrics/stats_counters.h"
#include "base/posix/eintr_wrapper.h"
#include "build/build_config.h"
#include "net/base/address_list.h"
#include "net/base/connection_type_histograms.h"
#include "net/base/io_buffer.h"
......@@ -24,6 +20,7 @@
#include "net/base/net_errors.h"
#include "net/base/net_util.h"
#include "net/base/network_change_notifier.h"
#include "net/socket/socket_libevent.h"
#include "net/socket/socket_net_log_params.h"
// If we don't have a definition for TCPI_OPT_SYN_DATA, create one.
......@@ -72,90 +69,15 @@ bool SetTCPKeepAlive(int fd, bool enable, int delay) {
return true;
}
int MapAcceptError(int os_error) {
switch (os_error) {
// If the client aborts the connection before the server calls accept,
// POSIX specifies accept should fail with ECONNABORTED. The server can
// ignore the error and just call accept again, so we map the error to
// ERR_IO_PENDING. See UNIX Network Programming, Vol. 1, 3rd Ed., Sec.
// 5.11, "Connection Abort before accept Returns".
case ECONNABORTED:
return ERR_IO_PENDING;
default:
return MapSystemError(os_error);
}
}
int MapConnectError(int os_error) {
switch (os_error) {
case EACCES:
return ERR_NETWORK_ACCESS_DENIED;
case ETIMEDOUT:
return ERR_CONNECTION_TIMED_OUT;
default: {
int net_error = MapSystemError(os_error);
if (net_error == ERR_FAILED)
return ERR_CONNECTION_FAILED; // More specific than ERR_FAILED.
// Give a more specific error when the user is offline.
if (net_error == ERR_ADDRESS_UNREACHABLE &&
NetworkChangeNotifier::IsOffline()) {
return ERR_INTERNET_DISCONNECTED;
}
return net_error;
}
}
}
} // namespace
//-----------------------------------------------------------------------------
TCPSocketLibevent::Watcher::Watcher(
const base::Closure& read_ready_callback,
const base::Closure& write_ready_callback)
: read_ready_callback_(read_ready_callback),
write_ready_callback_(write_ready_callback) {
}
TCPSocketLibevent::Watcher::~Watcher() {
}
void TCPSocketLibevent::Watcher::OnFileCanReadWithoutBlocking(int /* fd */) {
if (!read_ready_callback_.is_null())
read_ready_callback_.Run();
else
NOTREACHED();
}
void TCPSocketLibevent::Watcher::OnFileCanWriteWithoutBlocking(int /* fd */) {
if (!write_ready_callback_.is_null())
write_ready_callback_.Run();
else
NOTREACHED();
}
TCPSocketLibevent::TCPSocketLibevent(NetLog* net_log,
const NetLog::Source& source)
: socket_(kInvalidSocket),
accept_watcher_(base::Bind(&TCPSocketLibevent::DidCompleteAccept,
base::Unretained(this)),
base::Closure()),
accept_socket_(NULL),
accept_address_(NULL),
read_watcher_(base::Bind(&TCPSocketLibevent::DidCompleteRead,
base::Unretained(this)),
base::Closure()),
write_watcher_(base::Closure(),
base::Bind(&TCPSocketLibevent::DidCompleteConnectOrWrite,
base::Unretained(this))),
read_buf_len_(0),
write_buf_len_(0),
use_tcp_fastopen_(IsTCPFastOpenEnabled()),
: use_tcp_fastopen_(IsTCPFastOpenEnabled()),
tcp_fastopen_connected_(false),
fast_open_status_(FAST_OPEN_STATUS_UNKNOWN),
waiting_connect_(false),
connect_os_error_(0),
logging_multiple_connect_attempts_(false),
net_log_(BoundNetLog::Make(net_log, NetLog::SOURCE_SOCKET)) {
net_log_.BeginEvent(NetLog::TYPE_SOCKET_ALIVE,
......@@ -168,270 +90,167 @@ TCPSocketLibevent::~TCPSocketLibevent() {
UMA_HISTOGRAM_ENUMERATION("Net.TcpFastOpenSocketConnection",
fast_open_status_, FAST_OPEN_MAX_VALUE);
}
Close();
}
int TCPSocketLibevent::Open(AddressFamily family) {
DCHECK(CalledOnValidThread());
DCHECK_EQ(socket_, kInvalidSocket);
socket_ = CreatePlatformSocket(ConvertAddressFamily(family), SOCK_STREAM,
IPPROTO_TCP);
if (socket_ < 0) {
PLOG(ERROR) << "CreatePlatformSocket() returned an error";
return MapSystemError(errno);
}
if (SetNonBlocking(socket_)) {
int result = MapSystemError(errno);
Close();
return result;
}
return OK;
DCHECK(!socket_);
socket_.reset(new SocketLibevent);
int rv = socket_->Open(ConvertAddressFamily(family));
if (rv != OK)
socket_.reset();
return rv;
}
int TCPSocketLibevent::AdoptConnectedSocket(int socket,
int TCPSocketLibevent::AdoptConnectedSocket(int socket_fd,
const IPEndPoint& peer_address) {
DCHECK(CalledOnValidThread());
DCHECK_EQ(socket_, kInvalidSocket);
socket_ = socket;
DCHECK(!socket_);
if (SetNonBlocking(socket_)) {
int result = MapSystemError(errno);
Close();
return result;
SockaddrStorage storage;
if (!peer_address.ToSockAddr(storage.addr, &storage.addr_len) &&
// For backward compatibility, allows the empty address.
!(peer_address == IPEndPoint())) {
return ERR_ADDRESS_INVALID;
}
peer_address_.reset(new IPEndPoint(peer_address));
return OK;
socket_.reset(new SocketLibevent);
int rv = socket_->AdoptConnectedSocket(socket_fd, storage);
if (rv != OK)
socket_.reset();
return rv;
}
int TCPSocketLibevent::Bind(const IPEndPoint& address) {
DCHECK(CalledOnValidThread());
DCHECK_NE(socket_, kInvalidSocket);
DCHECK(socket_);
SockaddrStorage storage;
if (!address.ToSockAddr(storage.addr, &storage.addr_len))
return ERR_ADDRESS_INVALID;
int result = bind(socket_, storage.addr, storage.addr_len);
if (result < 0) {
PLOG(ERROR) << "bind() returned an error";
return MapSystemError(errno);
}
return OK;
return socket_->Bind(storage);
}
int TCPSocketLibevent::Listen(int backlog) {
DCHECK(CalledOnValidThread());
DCHECK_GT(backlog, 0);
DCHECK_NE(socket_, kInvalidSocket);
int result = listen(socket_, backlog);
if (result < 0) {
PLOG(ERROR) << "listen() returned an error";
return MapSystemError(errno);
}
return OK;
DCHECK(socket_);
return socket_->Listen(backlog);
}
int TCPSocketLibevent::Accept(scoped_ptr<TCPSocketLibevent>* socket,
int TCPSocketLibevent::Accept(scoped_ptr<TCPSocketLibevent>* tcp_socket,
IPEndPoint* address,
const CompletionCallback& callback) {
DCHECK(CalledOnValidThread());
DCHECK(socket);
DCHECK(address);
DCHECK(tcp_socket);
DCHECK(!callback.is_null());
DCHECK(accept_callback_.is_null());
DCHECK(socket_);
DCHECK(!accept_socket_);
net_log_.BeginEvent(NetLog::TYPE_TCP_ACCEPT);
int result = AcceptInternal(socket, address);
if (result == ERR_IO_PENDING) {
if (!base::MessageLoopForIO::current()->WatchFileDescriptor(
socket_, true, base::MessageLoopForIO::WATCH_READ,
&accept_socket_watcher_, &accept_watcher_)) {
PLOG(ERROR) << "WatchFileDescriptor failed on read";
return MapSystemError(errno);
}
accept_socket_ = socket;
accept_address_ = address;
accept_callback_ = callback;
}
return result;
int rv = socket_->Accept(
&accept_socket_,
base::Bind(&TCPSocketLibevent::AcceptCompleted,
base::Unretained(this), tcp_socket, address, callback));
if (rv != ERR_IO_PENDING)
rv = HandleAcceptCompleted(tcp_socket, address, rv);
return rv;
}
int TCPSocketLibevent::Connect(const IPEndPoint& address,
const CompletionCallback& callback) {
DCHECK(CalledOnValidThread());
DCHECK_NE(socket_, kInvalidSocket);
DCHECK(!waiting_connect_);
// |peer_address_| will be non-NULL if Connect() has been called. Unless
// Close() is called to reset the internal state, a second call to Connect()
// is not allowed.
// Please note that we don't allow a second Connect() even if the previous
// Connect() has failed. Connecting the same |socket_| again after a
// connection attempt failed results in unspecified behavior according to
// POSIX.
DCHECK(!peer_address_);
DCHECK(socket_);
if (!logging_multiple_connect_attempts_)
LogConnectBegin(AddressList(address));
peer_address_.reset(new IPEndPoint(address));
net_log_.BeginEvent(NetLog::TYPE_TCP_CONNECT_ATTEMPT,
CreateNetLogIPEndPointCallback(&address));
int rv = DoConnect();
if (rv == ERR_IO_PENDING) {
// Synchronous operation not supported.
DCHECK(!callback.is_null());
write_callback_ = callback;
waiting_connect_ = true;
} else {
DoConnectComplete(rv);
SockaddrStorage storage;
if (!address.ToSockAddr(storage.addr, &storage.addr_len))
return ERR_ADDRESS_INVALID;
if (use_tcp_fastopen_) {
// With TCP FastOpen, we pretend that the socket is connected.
DCHECK(!tcp_fastopen_connected_);
socket_->SetPeerAddress(storage);
return OK;
}
int rv = socket_->Connect(storage,
base::Bind(&TCPSocketLibevent::ConnectCompleted,
base::Unretained(this), callback));
if (rv != ERR_IO_PENDING)
rv = HandleConnectCompleted(rv);
return rv;
}
bool TCPSocketLibevent::IsConnected() const {
DCHECK(CalledOnValidThread());
if (socket_ == kInvalidSocket || waiting_connect_)
if (!socket_)
return false;
if (use_tcp_fastopen_ && !tcp_fastopen_connected_ && peer_address_) {
if (use_tcp_fastopen_ && !tcp_fastopen_connected_ &&
socket_->HasPeerAddress()) {
// With TCP FastOpen, we pretend that the socket is connected.
// This allows GetPeerAddress() to return peer_address_.
return true;
}
// Check if connection is alive.
char c;
int rv = HANDLE_EINTR(recv(socket_, &c, 1, MSG_PEEK));
if (rv == 0)
return false;
if (rv == -1 && errno != EAGAIN && errno != EWOULDBLOCK)
return false;
return true;
return socket_->IsConnected();
}
bool TCPSocketLibevent::IsConnectedAndIdle() const {
DCHECK(CalledOnValidThread());
if (socket_ == kInvalidSocket || waiting_connect_)
return false;
// TODO(wtc): should we also handle the TCP FastOpen case here,
// as we do in IsConnected()?
// Check if connection is alive and we haven't received any data
// unexpectedly.
char c;
int rv = HANDLE_EINTR(recv(socket_, &c, 1, MSG_PEEK));
if (rv >= 0)
return false;
if (errno != EAGAIN && errno != EWOULDBLOCK)
return false;
return true;
return socket_ && socket_->IsConnectedAndIdle();
}
int TCPSocketLibevent::Read(IOBuffer* buf,
int buf_len,
const CompletionCallback& callback) {
DCHECK(CalledOnValidThread());
DCHECK_NE(kInvalidSocket, socket_);
DCHECK(!waiting_connect_);
DCHECK(read_callback_.is_null());
// Synchronous operation not supported
DCHECK(socket_);
DCHECK(!callback.is_null());
DCHECK_GT(buf_len, 0);
int nread = HANDLE_EINTR(read(socket_, buf->data(), buf_len));
if (nread >= 0) {
base::StatsCounter read_bytes("tcp.read_bytes");
read_bytes.Add(nread);
net_log_.AddByteTransferEvent(NetLog::TYPE_SOCKET_BYTES_RECEIVED, nread,
buf->data());
int rv = socket_->Read(
buf, buf_len,
base::Bind(&TCPSocketLibevent::ReadCompleted,
base::Unretained(this), base::Unretained(buf), callback));
if (rv >= 0)
RecordFastOpenStatus();
return nread;
}
if (errno != EAGAIN && errno != EWOULDBLOCK) {
int net_error = MapSystemError(errno);
net_log_.AddEvent(NetLog::TYPE_SOCKET_READ_ERROR,
CreateNetLogSocketErrorCallback(net_error, errno));
return net_error;
}
if (!base::MessageLoopForIO::current()->WatchFileDescriptor(
socket_, true, base::MessageLoopForIO::WATCH_READ,
&read_socket_watcher_, &read_watcher_)) {
DVLOG(1) << "WatchFileDescriptor failed on read, errno " << errno;
return MapSystemError(errno);
}
read_buf_ = buf;
read_buf_len_ = buf_len;
read_callback_ = callback;
return ERR_IO_PENDING;
if (rv != ERR_IO_PENDING)
rv = HandleReadCompleted(buf, rv);
return rv;
}
int TCPSocketLibevent::Write(IOBuffer* buf,
int buf_len,
const CompletionCallback& callback) {
DCHECK(CalledOnValidThread());
DCHECK_NE(kInvalidSocket, socket_);
DCHECK(!waiting_connect_);
DCHECK(write_callback_.is_null());
// Synchronous operation not supported
DCHECK(socket_);
DCHECK(!callback.is_null());
DCHECK_GT(buf_len, 0);
int nwrite = InternalWrite(buf, buf_len);
if (nwrite >= 0) {
base::StatsCounter write_bytes("tcp.write_bytes");
write_bytes.Add(nwrite);
net_log_.AddByteTransferEvent(NetLog::TYPE_SOCKET_BYTES_SENT, nwrite,
buf->data());
return nwrite;
}
if (errno != EAGAIN && errno != EWOULDBLOCK) {
int net_error = MapSystemError(errno);
net_log_.AddEvent(NetLog::TYPE_SOCKET_WRITE_ERROR,
CreateNetLogSocketErrorCallback(net_error, errno));
return net_error;
}
if (!base::MessageLoopForIO::current()->WatchFileDescriptor(
socket_, true, base::MessageLoopForIO::WATCH_WRITE,
&write_socket_watcher_, &write_watcher_)) {
DVLOG(1) << "WatchFileDescriptor failed on write, errno " << errno;
return MapSystemError(errno);
CompletionCallback write_callback =
base::Bind(&TCPSocketLibevent::WriteCompleted,
base::Unretained(this), base::Unretained(buf), callback);
int rv;
if (use_tcp_fastopen_ && !tcp_fastopen_connected_) {
rv = TcpFastOpenWrite(buf, buf_len, write_callback);
} else {
rv = socket_->Write(buf, buf_len, write_callback);
}
write_buf_ = buf;
write_buf_len_ = buf_len;
write_callback_ = callback;
return ERR_IO_PENDING;
if (rv != ERR_IO_PENDING)
rv = HandleWriteCompleted(buf, rv);
return rv;
}
int TCPSocketLibevent::GetLocalAddress(IPEndPoint* address) const {
DCHECK(CalledOnValidThread());
DCHECK(address);
if (!socket_)
return ERR_SOCKET_NOT_CONNECTED;
SockaddrStorage storage;
if (getsockname(socket_, storage.addr, &storage.addr_len) < 0)
return MapSystemError(errno);
int rv = socket_->GetLocalAddress(&storage);
if (rv != OK)
return rv;
if (!address->FromSockAddr(storage.addr, storage.addr_len))
return ERR_ADDRESS_INVALID;
......@@ -439,25 +258,34 @@ int TCPSocketLibevent::GetLocalAddress(IPEndPoint* address) const {
}
int TCPSocketLibevent::GetPeerAddress(IPEndPoint* address) const {
DCHECK(CalledOnValidThread());
DCHECK(address);
if (!IsConnected())
return ERR_SOCKET_NOT_CONNECTED;
*address = *peer_address_;
SockaddrStorage storage;
int rv = socket_->GetPeerAddress(&storage);
if (rv != OK)
return rv;
if (!address->FromSockAddr(storage.addr, storage.addr_len))
return ERR_ADDRESS_INVALID;
return OK;
}
int TCPSocketLibevent::SetDefaultOptionsForServer() {
DCHECK(CalledOnValidThread());
DCHECK(socket_);
return SetAddressReuse(true);
}
void TCPSocketLibevent::SetDefaultOptionsForClient() {
DCHECK(CalledOnValidThread());
DCHECK(socket_);
// This mirrors the behaviour on Windows. See the comment in
// tcp_socket_win.cc after searching for "NODELAY".
SetTCPNoDelay(socket_, true); // If SetTCPNoDelay fails, we don't care.
// If SetTCPNoDelay fails, we don't care.
SetTCPNoDelay(socket_->socket_fd(), true);
// TCP keep alive wakes up the radio, which is expensive on mobile. Do not
// enable it there. It's useful to prevent TCP middleboxes from timing out
......@@ -473,12 +301,12 @@ void TCPSocketLibevent::SetDefaultOptionsForClient() {
#if !defined(OS_ANDROID) && !defined(OS_IOS)
const int kTCPKeepAliveSeconds = 45;
SetTCPKeepAlive(socket_, true, kTCPKeepAliveSeconds);
SetTCPKeepAlive(socket_->socket_fd(), true, kTCPKeepAliveSeconds);
#endif
}
int TCPSocketLibevent::SetAddressReuse(bool allow) {
DCHECK(CalledOnValidThread());
DCHECK(socket_);
// SO_REUSEADDR is useful for server sockets to bind to a recently unbound
// port. When a socket is closed, the end point changes its state to TIME_WAIT
......@@ -494,82 +322,51 @@ int TCPSocketLibevent::SetAddressReuse(bool allow) {
//
// SO_REUSEPORT is provided in MacOS X and iOS.
int boolean_value = allow ? 1 : 0;
int rv = setsockopt(socket_, SOL_SOCKET, SO_REUSEADDR, &boolean_value,
sizeof(boolean_value));
int rv = setsockopt(socket_->socket_fd(), SOL_SOCKET, SO_REUSEADDR,
&boolean_value, sizeof(boolean_value));
if (rv < 0)
return MapSystemError(errno);
return OK;
}
int TCPSocketLibevent::SetReceiveBufferSize(int32 size) {
DCHECK(CalledOnValidThread());
int rv = setsockopt(socket_, SOL_SOCKET, SO_RCVBUF,
DCHECK(socket_);
int rv = setsockopt(socket_->socket_fd(), SOL_SOCKET, SO_RCVBUF,
reinterpret_cast<const char*>(&size), sizeof(size));
return (rv == 0) ? OK : MapSystemError(errno);
}
int TCPSocketLibevent::SetSendBufferSize(int32 size) {
DCHECK(CalledOnValidThread());
int rv = setsockopt(socket_, SOL_SOCKET, SO_SNDBUF,
DCHECK(socket_);
int rv = setsockopt(socket_->socket_fd(), SOL_SOCKET, SO_SNDBUF,
reinterpret_cast<const char*>(&size), sizeof(size));
return (rv == 0) ? OK : MapSystemError(errno);
}
bool TCPSocketLibevent::SetKeepAlive(bool enable, int delay) {
DCHECK(CalledOnValidThread());
return SetTCPKeepAlive(socket_, enable, delay);
DCHECK(socket_);
return SetTCPKeepAlive(socket_->socket_fd(), enable, delay);
}
bool TCPSocketLibevent::SetNoDelay(bool no_delay) {
DCHECK(CalledOnValidThread());
return SetTCPNoDelay(socket_, no_delay);
DCHECK(socket_);
return SetTCPNoDelay(socket_->socket_fd(), no_delay);
}
void TCPSocketLibevent::Close() {
DCHECK(CalledOnValidThread());
bool ok = accept_socket_watcher_.StopWatchingFileDescriptor();
DCHECK(ok);
ok = read_socket_watcher_.StopWatchingFileDescriptor();
DCHECK(ok);
ok = write_socket_watcher_.StopWatchingFileDescriptor();
DCHECK(ok);
if (socket_ != kInvalidSocket) {
if (IGNORE_EINTR(close(socket_)) < 0)
PLOG(ERROR) << "close";
socket_ = kInvalidSocket;
}
if (!accept_callback_.is_null()) {
accept_socket_ = NULL;
accept_address_ = NULL;
accept_callback_.Reset();
}
if (!read_callback_.is_null()) {
read_buf_ = NULL;
read_buf_len_ = 0;
read_callback_.Reset();
}
if (!write_callback_.is_null()) {
write_buf_ = NULL;
write_buf_len_ = 0;
write_callback_.Reset();
}
socket_.reset();
tcp_fastopen_connected_ = false;
fast_open_status_ = FAST_OPEN_STATUS_UNKNOWN;
waiting_connect_ = false;
peer_address_.reset();
connect_os_error_ = 0;
}
bool TCPSocketLibevent::UsingTCPFastOpen() const {
return use_tcp_fastopen_;
}
bool TCPSocketLibevent::IsValid() const {
return socket_ != NULL && socket_->socket_fd() != kInvalidSocket;
}
void TCPSocketLibevent::StartLoggingMultipleConnectAttempts(
const AddressList& addresses) {
if (!logging_multiple_connect_attempts_) {
......@@ -589,98 +386,76 @@ void TCPSocketLibevent::EndLoggingMultipleConnectAttempts(int net_error) {
}
}
int TCPSocketLibevent::AcceptInternal(scoped_ptr<TCPSocketLibevent>* socket,
IPEndPoint* address) {
SockaddrStorage storage;
int new_socket = HANDLE_EINTR(accept(socket_,
storage.addr,
&storage.addr_len));
if (new_socket < 0) {
int net_error = MapAcceptError(errno);
if (net_error != ERR_IO_PENDING)
net_log_.EndEventWithNetErrorCode(NetLog::TYPE_TCP_ACCEPT, net_error);
return net_error;
}
void TCPSocketLibevent::AcceptCompleted(
scoped_ptr<TCPSocketLibevent>* tcp_socket,
IPEndPoint* address,
const CompletionCallback& callback,
int rv) {
DCHECK_NE(ERR_IO_PENDING, rv);
callback.Run(HandleAcceptCompleted(tcp_socket, address, rv));
}
IPEndPoint ip_end_point;
if (!ip_end_point.FromSockAddr(storage.addr, storage.addr_len)) {
NOTREACHED();
if (IGNORE_EINTR(close(new_socket)) < 0)
PLOG(ERROR) << "close";
int net_error = ERR_ADDRESS_INVALID;
net_log_.EndEventWithNetErrorCode(NetLog::TYPE_TCP_ACCEPT, net_error);
return net_error;
}
scoped_ptr<TCPSocketLibevent> tcp_socket(new TCPSocketLibevent(
net_log_.net_log(), net_log_.source()));
int adopt_result = tcp_socket->AdoptConnectedSocket(new_socket, ip_end_point);
if (adopt_result != OK) {
net_log_.EndEventWithNetErrorCode(NetLog::TYPE_TCP_ACCEPT, adopt_result);
return adopt_result;
}
*socket = tcp_socket.Pass();
*address = ip_end_point;
int TCPSocketLibevent::HandleAcceptCompleted(
scoped_ptr<TCPSocketLibevent>* tcp_socket,
IPEndPoint* address,
int rv) {
if (rv == OK)
rv = BuildTcpSocketLibevent(tcp_socket, address);
if (rv == OK) {
net_log_.EndEvent(NetLog::TYPE_TCP_ACCEPT,
CreateNetLogIPEndPointCallback(&ip_end_point));
return OK;
}
CreateNetLogIPEndPointCallback(address));
} else {
net_log_.EndEventWithNetErrorCode(NetLog::TYPE_TCP_ACCEPT, rv);
}
int TCPSocketLibevent::DoConnect() {
DCHECK_EQ(0, connect_os_error_);
return rv;
}
net_log_.BeginEvent(NetLog::TYPE_TCP_CONNECT_ATTEMPT,
CreateNetLogIPEndPointCallback(peer_address_.get()));
int TCPSocketLibevent::BuildTcpSocketLibevent(
scoped_ptr<TCPSocketLibevent>* tcp_socket,
IPEndPoint* address) {
DCHECK(accept_socket_);
// Connect the socket.
if (!use_tcp_fastopen_) {
SockaddrStorage storage;
if (!peer_address_->ToSockAddr(storage.addr, &storage.addr_len))
if (accept_socket_->GetPeerAddress(&storage) != OK ||
!address->FromSockAddr(storage.addr, storage.addr_len)) {
accept_socket_.reset();
return ERR_ADDRESS_INVALID;
if (!HANDLE_EINTR(connect(socket_, storage.addr, storage.addr_len))) {
// Connected without waiting!
return OK;
}
} else {
// With TCP FastOpen, we pretend that the socket is connected.
DCHECK(!tcp_fastopen_connected_);
return OK;
}
// Check if the connect() failed synchronously.
connect_os_error_ = errno;
if (connect_os_error_ != EINPROGRESS)
return MapConnectError(connect_os_error_);
// Otherwise the connect() is going to complete asynchronously, so watch
// for its completion.
if (!base::MessageLoopForIO::current()->WatchFileDescriptor(
socket_, true, base::MessageLoopForIO::WATCH_WRITE,
&write_socket_watcher_, &write_watcher_)) {
connect_os_error_ = errno;
DVLOG(1) << "WatchFileDescriptor failed: " << connect_os_error_;
return MapSystemError(connect_os_error_);
}
tcp_socket->reset(new TCPSocketLibevent(net_log_.net_log(),
net_log_.source()));
(*tcp_socket)->socket_.reset(accept_socket_.release());
return OK;
}
return ERR_IO_PENDING;
void TCPSocketLibevent::ConnectCompleted(const CompletionCallback& callback,
int rv) const {
DCHECK_NE(ERR_IO_PENDING, rv);
callback.Run(HandleConnectCompleted(rv));
}
void TCPSocketLibevent::DoConnectComplete(int result) {
int TCPSocketLibevent::HandleConnectCompleted(int rv) const {
// Log the end of this attempt (and any OS error it threw).
int os_error = connect_os_error_;
connect_os_error_ = 0;
if (result != OK) {
if (rv != OK) {
net_log_.EndEvent(NetLog::TYPE_TCP_CONNECT_ATTEMPT,
NetLog::IntegerCallback("os_error", os_error));
NetLog::IntegerCallback("os_error", errno));
} else {
net_log_.EndEvent(NetLog::TYPE_TCP_CONNECT_ATTEMPT);
}
// Give a more specific error when the user is offline.
if (rv == ERR_ADDRESS_UNREACHABLE && NetworkChangeNotifier::IsOffline())
rv = ERR_INTERNET_DISCONNECTED;
if (!logging_multiple_connect_attempts_)
LogConnectEnd(result);
LogConnectEnd(rv);
return rv;
}
void TCPSocketLibevent::LogConnectBegin(const AddressList& addresses) {
void TCPSocketLibevent::LogConnectBegin(const AddressList& addresses) const {
base::StatsCounter connects("tcp.connect");
connects.Increment();
......@@ -688,19 +463,18 @@ void TCPSocketLibevent::LogConnectBegin(const AddressList& addresses) {
addresses.CreateNetLogCallback());
}
void TCPSocketLibevent::LogConnectEnd(int net_error) {
if (net_error == OK)
UpdateConnectionTypeHistograms(CONNECTION_ANY);
void TCPSocketLibevent::LogConnectEnd(int net_error) const {
if (net_error != OK) {
net_log_.EndEventWithNetErrorCode(NetLog::TYPE_TCP_CONNECT, net_error);
return;
}
UpdateConnectionTypeHistograms(CONNECTION_ANY);
SockaddrStorage storage;
int rv = getsockname(socket_, storage.addr, &storage.addr_len);
if (rv != 0) {
PLOG(ERROR) << "getsockname() [rv: " << rv << "] error: ";
int rv = socket_->GetLocalAddress(&storage);
if (rv != OK) {
PLOG(ERROR) << "GetLocalAddress() [rv: " << rv << "] error: ";
NOTREACHED();
net_log_.EndEventWithNetErrorCode(NetLog::TYPE_TCP_CONNECT, rv);
return;
......@@ -711,121 +485,60 @@ void TCPSocketLibevent::LogConnectEnd(int net_error) {
storage.addr_len));
}
void TCPSocketLibevent::DidCompleteRead() {
void TCPSocketLibevent::ReadCompleted(IOBuffer* buf,
const CompletionCallback& callback,
int rv) {
DCHECK_NE(ERR_IO_PENDING, rv);
// Records fast open status regardless of error in asynchronous case.
// TODO(rdsmith,jri): Change histogram name to indicate it could be called on
// error.
RecordFastOpenStatus();
if (read_callback_.is_null())
return;
int bytes_transferred;
bytes_transferred = HANDLE_EINTR(read(socket_, read_buf_->data(),
read_buf_len_));
callback.Run(HandleReadCompleted(buf, rv));
}
int result;
if (bytes_transferred >= 0) {
result = bytes_transferred;
base::StatsCounter read_bytes("tcp.read_bytes");
read_bytes.Add(bytes_transferred);
net_log_.AddByteTransferEvent(NetLog::TYPE_SOCKET_BYTES_RECEIVED, result,
read_buf_->data());
} else {
result = MapSystemError(errno);
if (result != ERR_IO_PENDING) {
int TCPSocketLibevent::HandleReadCompleted(IOBuffer* buf, int rv) {
if (rv < 0) {
net_log_.AddEvent(NetLog::TYPE_SOCKET_READ_ERROR,
CreateNetLogSocketErrorCallback(result, errno));
}
CreateNetLogSocketErrorCallback(rv, errno));
return rv;
}
if (result != ERR_IO_PENDING) {
read_buf_ = NULL;
read_buf_len_ = 0;
bool ok = read_socket_watcher_.StopWatchingFileDescriptor();
DCHECK(ok);
base::ResetAndReturn(&read_callback_).Run(result);
}
base::StatsCounter read_bytes("tcp.read_bytes");
read_bytes.Add(rv);
net_log_.AddByteTransferEvent(NetLog::TYPE_SOCKET_BYTES_RECEIVED, rv,
buf->data());
return rv;
}
void TCPSocketLibevent::DidCompleteWrite() {
if (write_callback_.is_null())
return;
int bytes_transferred;
bytes_transferred = HANDLE_EINTR(write(socket_, write_buf_->data(),
write_buf_len_));
int result;
if (bytes_transferred >= 0) {
result = bytes_transferred;
base::StatsCounter write_bytes("tcp.write_bytes");
write_bytes.Add(bytes_transferred);
net_log_.AddByteTransferEvent(NetLog::TYPE_SOCKET_BYTES_SENT, result,
write_buf_->data());
} else {
result = MapSystemError(errno);
if (result != ERR_IO_PENDING) {
net_log_.AddEvent(NetLog::TYPE_SOCKET_WRITE_ERROR,
CreateNetLogSocketErrorCallback(result, errno));
}
}
if (result != ERR_IO_PENDING) {
write_buf_ = NULL;
write_buf_len_ = 0;
write_socket_watcher_.StopWatchingFileDescriptor();
base::ResetAndReturn(&write_callback_).Run(result);
}
void TCPSocketLibevent::WriteCompleted(IOBuffer* buf,
const CompletionCallback& callback,
int rv) const {
DCHECK_NE(ERR_IO_PENDING, rv);
callback.Run(HandleWriteCompleted(buf, rv));
}
void TCPSocketLibevent::DidCompleteConnect() {
DCHECK(waiting_connect_);
// Get the error that connect() completed with.
int os_error = 0;
socklen_t len = sizeof(os_error);
if (getsockopt(socket_, SOL_SOCKET, SO_ERROR, &os_error, &len) < 0)
os_error = errno;
int result = MapConnectError(os_error);
connect_os_error_ = os_error;
if (result != ERR_IO_PENDING) {
DoConnectComplete(result);
waiting_connect_ = false;
write_socket_watcher_.StopWatchingFileDescriptor();
base::ResetAndReturn(&write_callback_).Run(result);
int TCPSocketLibevent::HandleWriteCompleted(IOBuffer* buf, int rv) const {
if (rv < 0) {
net_log_.AddEvent(NetLog::TYPE_SOCKET_WRITE_ERROR,
CreateNetLogSocketErrorCallback(rv, errno));
return rv;
}
}
void TCPSocketLibevent::DidCompleteConnectOrWrite() {
if (waiting_connect_)
DidCompleteConnect();
else
DidCompleteWrite();
}
void TCPSocketLibevent::DidCompleteAccept() {
DCHECK(CalledOnValidThread());
int result = AcceptInternal(accept_socket_, accept_address_);
if (result != ERR_IO_PENDING) {
accept_socket_ = NULL;
accept_address_ = NULL;
bool ok = accept_socket_watcher_.StopWatchingFileDescriptor();
DCHECK(ok);
CompletionCallback callback = accept_callback_;
accept_callback_.Reset();
callback.Run(result);
}
base::StatsCounter write_bytes("tcp.write_bytes");
write_bytes.Add(rv);
net_log_.AddByteTransferEvent(NetLog::TYPE_SOCKET_BYTES_SENT, rv,
buf->data());
return rv;
}
int TCPSocketLibevent::InternalWrite(IOBuffer* buf, int buf_len) {
int nwrite;
if (use_tcp_fastopen_ && !tcp_fastopen_connected_) {
int TCPSocketLibevent::TcpFastOpenWrite(
IOBuffer* buf,
int buf_len,
const CompletionCallback& callback) {
SockaddrStorage storage;
if (!peer_address_->ToSockAddr(storage.addr, &storage.addr_len)) {
// Set errno to EADDRNOTAVAIL so that MapSystemError will map it to
// ERR_ADDRESS_INVALID later.
errno = EADDRNOTAVAIL;
return -1;
}
int rv = socket_->GetPeerAddress(&storage);
if (rv != OK)
return rv;
int flags = 0x20000000; // Magic flag to enable TCP_FASTOPEN.
#if defined(OS_LINUX)
......@@ -835,7 +548,7 @@ int TCPSocketLibevent::InternalWrite(IOBuffer* buf, int buf_len) {
// Open via sysctl.
flags |= MSG_NOSIGNAL;
#endif // defined(OS_LINUX)
nwrite = HANDLE_EINTR(sendto(socket_,
rv = HANDLE_EINTR(sendto(socket_->socket_fd(),
buf->data(),
buf_len,
flags,
......@@ -843,7 +556,11 @@ int TCPSocketLibevent::InternalWrite(IOBuffer* buf, int buf_len) {
storage.addr_len));
tcp_fastopen_connected_ = true;
if (nwrite < 0) {
if (rv >= 0) {
fast_open_status_ = FAST_OPEN_FAST_CONNECT_RETURN;
return rv;
}
DCHECK_NE(EPIPE, errno);
// If errno == EINPROGRESS, that means the kernel didn't have a cookie
......@@ -852,18 +569,18 @@ int TCPSocketLibevent::InternalWrite(IOBuffer* buf, int buf_len) {
// asynchronous cases. Note that the user buffer has not been copied to
// kernel space.
if (errno == EINPROGRESS) {
errno = EAGAIN;
fast_open_status_ = FAST_OPEN_SLOW_CONNECT_RETURN;
rv = ERR_IO_PENDING;
} else {
fast_open_status_ = FAST_OPEN_ERROR;
}
} else {
fast_open_status_ = FAST_OPEN_FAST_CONNECT_RETURN;
rv = MapSystemError(errno);
}
} else {
nwrite = HANDLE_EINTR(write(socket_, buf->data(), buf_len));
if (rv != ERR_IO_PENDING) {
fast_open_status_ = FAST_OPEN_ERROR;
return rv;
}
return nwrite;
fast_open_status_ = FAST_OPEN_SLOW_CONNECT_RETURN;
return socket_->WaitForWrite(buf, buf_len, callback);
}
void TCPSocketLibevent::RecordFastOpenStatus() {
......@@ -878,7 +595,8 @@ void TCPSocketLibevent::RecordFastOpenStatus() {
tcp_info info;
socklen_t info_len = sizeof(tcp_info);
getsockopt_success =
getsockopt(socket_, IPPROTO_TCP, TCP_INFO, &info, &info_len) == 0 &&
getsockopt(socket_->socket_fd(), IPPROTO_TCP, TCP_INFO,
&info, &info_len) == 0 &&
info_len == sizeof(tcp_info);
server_acked_data = getsockopt_success &&
(info.tcpi_options & TCPI_OPT_SYN_DATA);
......
......@@ -8,30 +8,27 @@
#include "base/basictypes.h"
#include "base/callback.h"
#include "base/compiler_specific.h"
#include "base/memory/ref_counted.h"
#include "base/memory/scoped_ptr.h"
#include "base/message_loop/message_loop.h"
#include "base/threading/non_thread_safe.h"
#include "net/base/address_family.h"
#include "net/base/completion_callback.h"
#include "net/base/net_export.h"
#include "net/base/net_log.h"
#include "net/socket/socket_descriptor.h"
namespace net {
class AddressList;
class IOBuffer;
class IPEndPoint;
class SocketLibevent;
class NET_EXPORT TCPSocketLibevent : public base::NonThreadSafe {
class NET_EXPORT TCPSocketLibevent {
public:
TCPSocketLibevent(NetLog* net_log, const NetLog::Source& source);
virtual ~TCPSocketLibevent();
int Open(AddressFamily family);
// Takes ownership of |socket|.
int AdoptConnectedSocket(int socket, const IPEndPoint& peer_address);
// Takes ownership of |socket_fd|.
int AdoptConnectedSocket(int socket_fd, const IPEndPoint& peer_address);
int Bind(const IPEndPoint& address);
......@@ -69,7 +66,7 @@ class NET_EXPORT TCPSocketLibevent : public base::NonThreadSafe {
void Close();
bool UsingTCPFastOpen() const;
bool IsValid() const { return socket_ != kInvalidSocket; }
bool IsValid() const;
// Marks the start/end of a series of connect attempts for logging purpose.
//
......@@ -136,76 +133,39 @@ class NET_EXPORT TCPSocketLibevent : public base::NonThreadSafe {
FAST_OPEN_MAX_VALUE
};
// Watcher simply forwards notifications to Closure objects set via the
// constructor.
class Watcher: public base::MessageLoopForIO::Watcher {
public:
Watcher(const base::Closure& read_ready_callback,
const base::Closure& write_ready_callback);
virtual ~Watcher();
// base::MessageLoopForIO::Watcher methods.
virtual void OnFileCanReadWithoutBlocking(int fd) OVERRIDE;
virtual void OnFileCanWriteWithoutBlocking(int fd) OVERRIDE;
private:
base::Closure read_ready_callback_;
base::Closure write_ready_callback_;
DISALLOW_COPY_AND_ASSIGN(Watcher);
};
int AcceptInternal(scoped_ptr<TCPSocketLibevent>* socket,
void AcceptCompleted(scoped_ptr<TCPSocketLibevent>* tcp_socket,
IPEndPoint* address,
const CompletionCallback& callback,
int rv);
int HandleAcceptCompleted(scoped_ptr<TCPSocketLibevent>* tcp_socket,
IPEndPoint* address,
int rv);
int BuildTcpSocketLibevent(scoped_ptr<TCPSocketLibevent>* tcp_socket,
IPEndPoint* address);
int DoConnect();
void DoConnectComplete(int result);
void LogConnectBegin(const AddressList& addresses);
void LogConnectEnd(int net_error);
void DidCompleteRead();
void DidCompleteWrite();
void DidCompleteConnect();
void DidCompleteConnectOrWrite();
void DidCompleteAccept();
// Internal function to write to a socket. Returns an OS error.
int InternalWrite(IOBuffer* buf, int buf_len);
void ConnectCompleted(const CompletionCallback& callback, int rv) const;
int HandleConnectCompleted(int rv) const;
void LogConnectBegin(const AddressList& addresses) const;
void LogConnectEnd(int net_error) const;
void ReadCompleted(IOBuffer* buf,
const CompletionCallback& callback,
int rv);
int HandleReadCompleted(IOBuffer* buf, int rv);
void WriteCompleted(IOBuffer* buf,
const CompletionCallback& callback,
int rv) const;
int HandleWriteCompleted(IOBuffer* buf, int rv) const;
int TcpFastOpenWrite(IOBuffer* buf,
int buf_len,
const CompletionCallback& callback);
// Called when the socket is known to be in a connected state.
void RecordFastOpenStatus();
int socket_;
base::MessageLoopForIO::FileDescriptorWatcher accept_socket_watcher_;
Watcher accept_watcher_;
scoped_ptr<TCPSocketLibevent>* accept_socket_;
IPEndPoint* accept_address_;
CompletionCallback accept_callback_;
// The socket's libevent wrappers for reads and writes.
base::MessageLoopForIO::FileDescriptorWatcher read_socket_watcher_;
base::MessageLoopForIO::FileDescriptorWatcher write_socket_watcher_;
// The corresponding watchers for reads and writes.
Watcher read_watcher_;
Watcher write_watcher_;
// The buffer used for reads.
scoped_refptr<IOBuffer> read_buf_;
int read_buf_len_;
// The buffer used for writes.
scoped_refptr<IOBuffer> write_buf_;
int write_buf_len_;
// External callback; called when read is complete.
CompletionCallback read_callback_;
// External callback; called when write or connect is complete.
CompletionCallback write_callback_;
scoped_ptr<SocketLibevent> socket_;
scoped_ptr<SocketLibevent> accept_socket_;
// Enables experimental TCP FastOpen option.
const bool use_tcp_fastopen_;
......@@ -215,14 +175,6 @@ class NET_EXPORT TCPSocketLibevent : public base::NonThreadSafe {
FastOpenStatus fast_open_status_;
// A connect operation is pending. In this case, |write_callback_| needs to be
// called when connect is complete.
bool waiting_connect_;
scoped_ptr<IPEndPoint> peer_address_;
// The OS error that a connect attempt last completed with.
int connect_os_error_;
bool logging_multiple_connect_attempts_;
BoundNetLog net_log_;
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment