net/socket/socket_libevent.cc

   1 // Copyright 2014 The Chromium Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style license that can be
   3 // found in the LICENSE file.
   4
   5 #include "net/socket/socket_libevent.h"
   6
   7 #include <errno.h>
   8 #include <netinet/in.h>
   9 #include <sys/socket.h>
  10
  11 #include "base/callback_helpers.h"
  12 #include "base/logging.h"
  13 #include "base/posix/eintr_wrapper.h"
  14 #include "net/base/io_buffer.h"
  15 #include "net/base/ip_endpoint.h"
  16 #include "net/base/net_errors.h"
  17 #include "net/base/net_util.h"
  18
  19 namespace net {
  20
  21 namespace {
  22
  23 int MapAcceptError(int os_error) {
  24   switch (os_error) {
  25     // If the client aborts the connection before the server calls accept,
  26     // POSIX specifies accept should fail with ECONNABORTED. The server can
  27     // ignore the error and just call accept again, so we map the error to
  28     // ERR_IO_PENDING. See UNIX Network Programming, Vol. 1, 3rd Ed., Sec.
  29     // 5.11, "Connection Abort before accept Returns".
  30     case ECONNABORTED:
  31       return ERR_IO_PENDING;
  32     default:
  33       return MapSystemError(os_error);
  34   }
  35 }
  36
  37 int MapConnectError(int os_error) {
  38   switch (os_error) {
  39     case EINPROGRESS:
  40       return ERR_IO_PENDING;
  41     case EACCES:
  42       return ERR_NETWORK_ACCESS_DENIED;
  43     case ETIMEDOUT:
  44       return ERR_CONNECTION_TIMED_OUT;
  45     default: {
  46       int net_error = MapSystemError(os_error);
  47       if (net_error == ERR_FAILED)
  48         return ERR_CONNECTION_FAILED;  // More specific than ERR_FAILED.
  49       return net_error;
  50     }
  51   }
  52 }
  53
  54 }  // namespace
  55
  56 SocketLibevent::SocketLibevent()
  57     : socket_fd_(kInvalidSocket),
  58       read_buf_len_(0),
  59       write_buf_len_(0),
  60       waiting_connect_(false) {
  61 }
  62
  63 SocketLibevent::~SocketLibevent() {
  64   Close();
  65 }
  66
  67 int SocketLibevent::Open(int address_family) {
  68   DCHECK(thread_checker_.CalledOnValidThread());
  69   DCHECK_EQ(kInvalidSocket, socket_fd_);
  70   DCHECK(address_family == AF_INET ||
  71          address_family == AF_INET6 ||
  72          address_family == AF_UNIX);
  73
  74   socket_fd_ = CreatePlatformSocket(
  75       address_family,
  76       SOCK_STREAM,
  77       address_family == AF_UNIX ? 0 : IPPROTO_TCP);
  78   if (socket_fd_ < 0) {
  79     PLOG(ERROR) << "CreatePlatformSocket() returned an error, errno=" << errno;
  80     return MapSystemError(errno);
  81   }
  82
  83   if (SetNonBlocking(socket_fd_)) {
  84     int rv = MapSystemError(errno);
  85     Close();
  86     return rv;
  87   }
  88
  89   return OK;
  90 }
  91
  92 int SocketLibevent::AdoptConnectedSocket(SocketDescriptor socket,
  93                                          const SockaddrStorage& address) {
  94   DCHECK(thread_checker_.CalledOnValidThread());
  95   DCHECK_EQ(kInvalidSocket, socket_fd_);
  96
  97   socket_fd_ = socket;
  98
  99   if (SetNonBlocking(socket_fd_)) {
 100     int rv = MapSystemError(errno);
 101     Close();
 102     return rv;
 103   }
 104
 105   SetPeerAddress(address);
 106   return OK;
 107 }
 108
 109 SocketDescriptor SocketLibevent::ReleaseConnectedSocket() {
 110   StopWatchingAndCleanUp();
 111   SocketDescriptor socket_fd = socket_fd_;
 112   socket_fd_ = kInvalidSocket;
 113   return socket_fd;
 114 }
 115
 116 int SocketLibevent::Bind(const SockaddrStorage& address) {
 117   DCHECK(thread_checker_.CalledOnValidThread());
 118   DCHECK_NE(kInvalidSocket, socket_fd_);
 119
 120   int rv = bind(socket_fd_, address.addr, address.addr_len);
 121   if (rv < 0) {
 122     PLOG(ERROR) << "bind() returned an error, errno=" << errno;
 123     return MapSystemError(errno);
 124   }
 125
 126   return OK;
 127 }
 128
 129 int SocketLibevent::Listen(int backlog) {
 130   DCHECK(thread_checker_.CalledOnValidThread());
 131   DCHECK_NE(kInvalidSocket, socket_fd_);
 132   DCHECK_LT(0, backlog);
 133
 134   int rv = listen(socket_fd_, backlog);
 135   if (rv < 0) {
 136     PLOG(ERROR) << "listen() returned an error, errno=" << errno;
 137     return MapSystemError(errno);
 138   }
 139
 140   return OK;
 141 }
 142
 143 int SocketLibevent::Accept(scoped_ptr<SocketLibevent>* socket,
 144                            const CompletionCallback& callback) {
 145   DCHECK(thread_checker_.CalledOnValidThread());
 146   DCHECK_NE(kInvalidSocket, socket_fd_);
 147   DCHECK(accept_callback_.is_null());
 148   DCHECK(socket);
 149   DCHECK(!callback.is_null());
 150
 151   int rv = DoAccept(socket);
 152   if (rv != ERR_IO_PENDING)
 153     return rv;
 154
 155   if (!base::MessageLoopForIO::current()->WatchFileDescriptor(
 156           socket_fd_, true, base::MessageLoopForIO::WATCH_READ,
 157           &accept_socket_watcher_, this)) {
 158     PLOG(ERROR) << "WatchFileDescriptor failed on accept, errno " << errno;
 159     return MapSystemError(errno);
 160   }
 161
 162   accept_socket_ = socket;
 163   accept_callback_ = callback;
 164   return ERR_IO_PENDING;
 165 }
 166
 167 int SocketLibevent::Connect(const SockaddrStorage& address,
 168                             const CompletionCallback& callback) {
 169   DCHECK(thread_checker_.CalledOnValidThread());
 170   DCHECK_NE(kInvalidSocket, socket_fd_);
 171   DCHECK(!waiting_connect_);
 172   DCHECK(!callback.is_null());
 173
 174   SetPeerAddress(address);
 175
 176   int rv = DoConnect();
 177   if (rv != ERR_IO_PENDING)
 178     return rv;
 179
 180   if (!base::MessageLoopForIO::current()->WatchFileDescriptor(
 181           socket_fd_, true, base::MessageLoopForIO::WATCH_WRITE,
 182           &write_socket_watcher_, this)) {
 183     PLOG(ERROR) << "WatchFileDescriptor failed on connect, errno " << errno;
 184     return MapSystemError(errno);
 185   }
 186
 187   write_callback_ = callback;
 188   waiting_connect_ = true;
 189   return ERR_IO_PENDING;
 190 }
 191
 192 bool SocketLibevent::IsConnected() const {
 193   DCHECK(thread_checker_.CalledOnValidThread());
 194
 195   if (socket_fd_ == kInvalidSocket || waiting_connect_)
 196     return false;
 197
 198   // Checks if connection is alive.
 199   char c;
 200   int rv = HANDLE_EINTR(recv(socket_fd_, &c, 1, MSG_PEEK));
 201   if (rv == 0)
 202     return false;
 203   if (rv == -1 && errno != EAGAIN && errno != EWOULDBLOCK)
 204     return false;
 205
 206   return true;
 207 }
 208
 209 bool SocketLibevent::IsConnectedAndIdle() const {
 210   DCHECK(thread_checker_.CalledOnValidThread());
 211
 212   if (socket_fd_ == kInvalidSocket || waiting_connect_)
 213     return false;
 214
 215   // Check if connection is alive and we haven't received any data
 216   // unexpectedly.
 217   char c;
 218   int rv = HANDLE_EINTR(recv(socket_fd_, &c, 1, MSG_PEEK));
 219   if (rv >= 0)
 220     return false;
 221   if (errno != EAGAIN && errno != EWOULDBLOCK)
 222     return false;
 223
 224   return true;
 225 }
 226
 227 int SocketLibevent::Read(IOBuffer* buf,
 228                          int buf_len,
 229                          const CompletionCallback& callback) {
 230   DCHECK(thread_checker_.CalledOnValidThread());
 231   DCHECK_NE(kInvalidSocket, socket_fd_);
 232   DCHECK(!waiting_connect_);
 233   DCHECK(read_callback_.is_null());
 234   // Synchronous operation not supported
 235   DCHECK(!callback.is_null());
 236   DCHECK_LT(0, buf_len);
 237
 238   int rv = DoRead(buf, buf_len);
 239   if (rv != ERR_IO_PENDING)
 240     return rv;
 241
 242   if (!base::MessageLoopForIO::current()->WatchFileDescriptor(
 243           socket_fd_, true, base::MessageLoopForIO::WATCH_READ,
 244           &read_socket_watcher_, this)) {
 245     PLOG(ERROR) << "WatchFileDescriptor failed on read, errno " << errno;
 246     return MapSystemError(errno);
 247   }
 248
 249   read_buf_ = buf;
 250   read_buf_len_ = buf_len;
 251   read_callback_ = callback;
 252   return ERR_IO_PENDING;
 253 }
 254
 255 int SocketLibevent::Write(IOBuffer* buf,
 256                           int buf_len,
 257                           const CompletionCallback& callback) {
 258   DCHECK(thread_checker_.CalledOnValidThread());
 259   DCHECK_NE(kInvalidSocket, socket_fd_);
 260   DCHECK(!waiting_connect_);
 261   DCHECK(write_callback_.is_null());
 262   // Synchronous operation not supported
 263   DCHECK(!callback.is_null());
 264   DCHECK_LT(0, buf_len);
 265
 266   int rv = DoWrite(buf, buf_len);
 267   if (rv == ERR_IO_PENDING)
 268     rv = WaitForWrite(buf, buf_len, callback);
 269   return rv;
 270 }
 271
 272 int SocketLibevent::WaitForWrite(IOBuffer* buf,
 273                                  int buf_len,
 274                                  const CompletionCallback& callback) {
 275   DCHECK(thread_checker_.CalledOnValidThread());
 276   DCHECK_NE(kInvalidSocket, socket_fd_);
 277   DCHECK(write_callback_.is_null());
 278   // Synchronous operation not supported
 279   DCHECK(!callback.is_null());
 280   DCHECK_LT(0, buf_len);
 281
 282   if (!base::MessageLoopForIO::current()->WatchFileDescriptor(
 283           socket_fd_, true, base::MessageLoopForIO::WATCH_WRITE,
 284           &write_socket_watcher_, this)) {
 285     PLOG(ERROR) << "WatchFileDescriptor failed on write, errno " << errno;
 286     return MapSystemError(errno);
 287   }
 288
 289   write_buf_ = buf;
 290   write_buf_len_ = buf_len;
 291   write_callback_ = callback;
 292   return ERR_IO_PENDING;
 293 }
 294
 295 int SocketLibevent::GetLocalAddress(SockaddrStorage* address) const {
 296   DCHECK(thread_checker_.CalledOnValidThread());
 297   DCHECK(address);
 298
 299   if (getsockname(socket_fd_, address->addr, &address->addr_len) < 0)
 300     return MapSystemError(errno);
 301   return OK;
 302 }
 303
 304 int SocketLibevent::GetPeerAddress(SockaddrStorage* address) const {
 305   DCHECK(thread_checker_.CalledOnValidThread());
 306   DCHECK(address);
 307
 308   if (!HasPeerAddress())
 309     return ERR_SOCKET_NOT_CONNECTED;
 310
 311   *address = *peer_address_;
 312   return OK;
 313 }
 314
 315 void SocketLibevent::SetPeerAddress(const SockaddrStorage& address) {
 316   DCHECK(thread_checker_.CalledOnValidThread());
 317   // |peer_address_| will be non-NULL if Connect() has been called. Unless
 318   // Close() is called to reset the internal state, a second call to Connect()
 319   // is not allowed.
 320   // Please note that we don't allow a second Connect() even if the previous
 321   // Connect() has failed. Connecting the same |socket_| again after a
 322   // connection attempt failed results in unspecified behavior according to
 323   // POSIX.
 324   DCHECK(!peer_address_);
 325   peer_address_.reset(new SockaddrStorage(address));
 326 }
 327
 328 bool SocketLibevent::HasPeerAddress() const {
 329   DCHECK(thread_checker_.CalledOnValidThread());
 330   return peer_address_ != NULL;
 331 }
 332
 333 void SocketLibevent::Close() {
 334   DCHECK(thread_checker_.CalledOnValidThread());
 335
 336   StopWatchingAndCleanUp();
 337
 338   if (socket_fd_ != kInvalidSocket) {
 339     if (IGNORE_EINTR(close(socket_fd_)) < 0)
 340       PLOG(ERROR) << "close() returned an error, errno=" << errno;
 341     socket_fd_ = kInvalidSocket;
 342   }
 343 }
 344
 345 void SocketLibevent::OnFileCanReadWithoutBlocking(int fd) {
 346   DCHECK(!accept_callback_.is_null() || !read_callback_.is_null());
 347   if (!accept_callback_.is_null()) {
 348     AcceptCompleted();
 349   } else {  // !read_callback_.is_null()
 350     ReadCompleted();
 351   }
 352 }
 353
 354 void SocketLibevent::OnFileCanWriteWithoutBlocking(int fd) {
 355   DCHECK(!write_callback_.is_null());
 356   if (waiting_connect_) {
 357     ConnectCompleted();
 358   } else {
 359     WriteCompleted();
 360   }
 361 }
 362
 363 int SocketLibevent::DoAccept(scoped_ptr<SocketLibevent>* socket) {
 364   SockaddrStorage new_peer_address;
 365   int new_socket = HANDLE_EINTR(accept(socket_fd_,
 366                                        new_peer_address.addr,
 367                                        &new_peer_address.addr_len));
 368   if (new_socket < 0)
 369     return MapAcceptError(errno);
 370
 371   scoped_ptr<SocketLibevent> accepted_socket(new SocketLibevent);
 372   int rv = accepted_socket->AdoptConnectedSocket(new_socket, new_peer_address);
 373   if (rv != OK)
 374     return rv;
 375
 376   *socket = accepted_socket.Pass();
 377   return OK;
 378 }
 379
 380 void SocketLibevent::AcceptCompleted() {
 381   DCHECK(accept_socket_);
 382   int rv = DoAccept(accept_socket_);
 383   if (rv == ERR_IO_PENDING)
 384     return;
 385
 386   bool ok = accept_socket_watcher_.StopWatchingFileDescriptor();
 387   DCHECK(ok);
 388   accept_socket_ = NULL;
 389   base::ResetAndReturn(&accept_callback_).Run(rv);
 390 }
 391
 392 int SocketLibevent::DoConnect() {
 393   int rv = HANDLE_EINTR(connect(socket_fd_,
 394                                 peer_address_->addr,
 395                                 peer_address_->addr_len));
 396   DCHECK_GE(0, rv);
 397   return rv == 0 ? OK : MapConnectError(errno);
 398 }
 399
 400 void SocketLibevent::ConnectCompleted() {
 401   // Get the error that connect() completed with.
 402   int os_error = 0;
 403   socklen_t len = sizeof(os_error);
 404   if (getsockopt(socket_fd_, SOL_SOCKET, SO_ERROR, &os_error, &len) == 0) {
 405     // TCPSocketLibevent expects errno to be set.
 406     errno = os_error;
 407   }
 408
 409   int rv = MapConnectError(errno);
 410   if (rv == ERR_IO_PENDING)
 411     return;
 412
 413   bool ok = write_socket_watcher_.StopWatchingFileDescriptor();
 414   DCHECK(ok);
 415   waiting_connect_ = false;
 416   base::ResetAndReturn(&write_callback_).Run(rv);
 417 }
 418
 419 int SocketLibevent::DoRead(IOBuffer* buf, int buf_len) {
 420   int rv = HANDLE_EINTR(read(socket_fd_, buf->data(), buf_len));
 421   return rv >= 0 ? rv : MapSystemError(errno);
 422 }
 423
 424 void SocketLibevent::ReadCompleted() {
 425   int rv = DoRead(read_buf_.get(), read_buf_len_);
 426   if (rv == ERR_IO_PENDING)
 427     return;
 428
 429   bool ok = read_socket_watcher_.StopWatchingFileDescriptor();
 430   DCHECK(ok);
 431   read_buf_ = NULL;
 432   read_buf_len_ = 0;
 433   base::ResetAndReturn(&read_callback_).Run(rv);
 434 }
 435
 436 int SocketLibevent::DoWrite(IOBuffer* buf, int buf_len) {
 437   int rv = HANDLE_EINTR(write(socket_fd_, buf->data(), buf_len));
 438   return rv >= 0 ? rv : MapSystemError(errno);
 439 }
 440
 441 void SocketLibevent::WriteCompleted() {
 442   int rv = DoWrite(write_buf_.get(), write_buf_len_);
 443   if (rv == ERR_IO_PENDING)
 444     return;
 445
 446   bool ok = write_socket_watcher_.StopWatchingFileDescriptor();
 447   DCHECK(ok);
 448   write_buf_ = NULL;
 449   write_buf_len_ = 0;
 450   base::ResetAndReturn(&write_callback_).Run(rv);
 451 }
 452
 453 void SocketLibevent::StopWatchingAndCleanUp() {
 454   bool ok = accept_socket_watcher_.StopWatchingFileDescriptor();
 455   DCHECK(ok);
 456   ok = read_socket_watcher_.StopWatchingFileDescriptor();
 457   DCHECK(ok);
 458   ok = write_socket_watcher_.StopWatchingFileDescriptor();
 459   DCHECK(ok);
 460
 461   if (!accept_callback_.is_null()) {
 462     accept_socket_ = NULL;
 463     accept_callback_.Reset();
 464   }
 465
 466   if (!read_callback_.is_null()) {
 467     read_buf_ = NULL;
 468     read_buf_len_ = 0;
 469     read_callback_.Reset();
 470   }
 471
 472   if (!write_callback_.is_null()) {
 473     write_buf_ = NULL;
 474     write_buf_len_ = 0;
 475     write_callback_.Reset();
 476   }
 477
 478   waiting_connect_ = false;
 479   peer_address_.reset();
 480 }
 481
 482 }  // namespace net