net/tools/quic/quic_client.cc

   1 // Copyright (c) 2012 The Chromium Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style license that can be
   3 // found in the LICENSE file.
   4
   5 #include "net/tools/quic/quic_client.h"
   6
   7 #include <errno.h>
   8 #include <netinet/in.h>
   9 #include <string.h>
  10 #include <sys/epoll.h>
  11 #include <sys/socket.h>
  12 #include <unistd.h>
  13
  14 #include "base/logging.h"
  15 #include "net/base/net_util.h"
  16 #include "net/quic/crypto/quic_random.h"
  17 #include "net/quic/quic_connection.h"
  18 #include "net/quic/quic_data_reader.h"
  19 #include "net/quic/quic_flags.h"
  20 #include "net/quic/quic_protocol.h"
  21 #include "net/quic/quic_server_id.h"
  22 #include "net/tools/quic/quic_epoll_connection_helper.h"
  23 #include "net/tools/quic/quic_socket_utils.h"
  24 #include "net/tools/quic/spdy_balsa_utils.h"
  25
  26 #ifndef SO_RXQ_OVFL
  27 #define SO_RXQ_OVFL 40
  28 #endif
  29
  30 using std::string;
  31 using std::vector;
  32
  33 namespace net {
  34 namespace tools {
  35
  36 const int kEpollFlags = EPOLLIN | EPOLLOUT | EPOLLET;
  37
  38 void QuicClient::ClientQuicDataToResend::Resend() {
  39   client_->SendRequest(*headers_, body_, fin_);
  40   delete headers_;
  41   headers_ = nullptr;
  42 }
  43
  44 QuicClient::QuicClient(IPEndPoint server_address,
  45                        const QuicServerId& server_id,
  46                        const QuicVersionVector& supported_versions,
  47                        EpollServer* epoll_server)
  48     : QuicClient(server_address,
  49                  server_id,
  50                  supported_versions,
  51                  QuicConfig(),
  52                  epoll_server) {}
  53
  54 QuicClient::QuicClient(IPEndPoint server_address,
  55                        const QuicServerId& server_id,
  56                        const QuicVersionVector& supported_versions,
  57                        const QuicConfig& config,
  58                        EpollServer* epoll_server)
  59     : QuicClientBase(server_id, supported_versions, config),
  60       server_address_(server_address),
  61       local_port_(0),
  62       epoll_server_(epoll_server),
  63       fd_(-1),
  64       helper_(CreateQuicConnectionHelper()),
  65       initialized_(false),
  66       packets_dropped_(0),
  67       overflow_supported_(false),
  68       store_response_(false),
  69       latest_response_code_(-1) {}
  70
  71 QuicClient::~QuicClient() {
  72   if (connected()) {
  73     session()->connection()->SendConnectionClose(QUIC_PEER_GOING_AWAY);
  74   }
  75
  76   STLDeleteElements(&data_to_resend_on_connect_);
  77   STLDeleteElements(&data_sent_before_handshake_);
  78
  79   CleanUpUDPSocketImpl();
  80 }
  81
  82 bool QuicClient::Initialize() {
  83   QuicClientBase::Initialize();
  84
  85   // If an initial flow control window has not explicitly been set, then use the
  86   // same values that Chrome uses.
  87   const uint32 kSessionMaxRecvWindowSize = 15 * 1024 * 1024;  // 15 MB
  88   const uint32 kStreamMaxRecvWindowSize = 6 * 1024 * 1024;    //  6 MB
  89   if (config()->GetInitialStreamFlowControlWindowToSend() ==
  90       kMinimumFlowControlSendWindow) {
  91     config()->SetInitialStreamFlowControlWindowToSend(kStreamMaxRecvWindowSize);
  92   }
  93   if (config()->GetInitialSessionFlowControlWindowToSend() ==
  94       kMinimumFlowControlSendWindow) {
  95     config()->SetInitialSessionFlowControlWindowToSend(
  96         kSessionMaxRecvWindowSize);
  97   }
  98
  99   epoll_server_->set_timeout_in_us(50 * 1000);
 100
 101   if (!CreateUDPSocket()) {
 102     return false;
 103   }
 104
 105   epoll_server_->RegisterFD(fd_, this, kEpollFlags);
 106   initialized_ = true;
 107   return true;
 108 }
 109
 110 QuicClient::QuicDataToResend::QuicDataToResend(BalsaHeaders* headers,
 111                                                StringPiece body,
 112                                                bool fin)
 113     : headers_(headers), body_(body), fin_(fin) {}
 114
 115 QuicClient::QuicDataToResend::~QuicDataToResend() {
 116   if (headers_) {
 117     delete headers_;
 118   }
 119 }
 120
 121 bool QuicClient::CreateUDPSocket() {
 122   int address_family = server_address_.GetSockAddrFamily();
 123   fd_ = socket(address_family, SOCK_DGRAM | SOCK_NONBLOCK, IPPROTO_UDP);
 124   if (fd_ < 0) {
 125     LOG(ERROR) << "CreateSocket() failed: " << strerror(errno);
 126     return false;
 127   }
 128
 129   int get_overflow = 1;
 130   int rc = setsockopt(fd_, SOL_SOCKET, SO_RXQ_OVFL, &get_overflow,
 131                       sizeof(get_overflow));
 132   if (rc < 0) {
 133     DLOG(WARNING) << "Socket overflow detection not supported";
 134   } else {
 135     overflow_supported_ = true;
 136   }
 137
 138   if (!QuicSocketUtils::SetReceiveBufferSize(fd_,
 139                                              kDefaultSocketReceiveBuffer)) {
 140     return false;
 141   }
 142
 143   if (!QuicSocketUtils::SetSendBufferSize(fd_, kDefaultSocketReceiveBuffer)) {
 144     return false;
 145   }
 146
 147   rc = QuicSocketUtils::SetGetAddressInfo(fd_, address_family);
 148   if (rc < 0) {
 149     LOG(ERROR) << "IP detection not supported" << strerror(errno);
 150     return false;
 151   }
 152
 153   if (bind_to_address_.size() != 0) {
 154     client_address_ = IPEndPoint(bind_to_address_, local_port_);
 155   } else if (address_family == AF_INET) {
 156     IPAddressNumber any4;
 157     CHECK(net::ParseIPLiteralToNumber("0.0.0.0", &any4));
 158     client_address_ = IPEndPoint(any4, local_port_);
 159   } else {
 160     IPAddressNumber any6;
 161     CHECK(net::ParseIPLiteralToNumber("::", &any6));
 162     client_address_ = IPEndPoint(any6, local_port_);
 163   }
 164
 165   sockaddr_storage raw_addr;
 166   socklen_t raw_addr_len = sizeof(raw_addr);
 167   CHECK(client_address_.ToSockAddr(reinterpret_cast<sockaddr*>(&raw_addr),
 168                            &raw_addr_len));
 169   rc = bind(fd_,
 170             reinterpret_cast<const sockaddr*>(&raw_addr),
 171             sizeof(raw_addr));
 172   if (rc < 0) {
 173     LOG(ERROR) << "Bind failed: " << strerror(errno);
 174     return false;
 175   }
 176
 177   SockaddrStorage storage;
 178   if (getsockname(fd_, storage.addr, &storage.addr_len) != 0 ||
 179       !client_address_.FromSockAddr(storage.addr, storage.addr_len)) {
 180     LOG(ERROR) << "Unable to get self address.  Error: " << strerror(errno);
 181   }
 182
 183   return true;
 184 }
 185
 186 bool QuicClient::Connect() {
 187   // Attempt multiple connects until the maximum number of client hellos have
 188   // been sent.
 189   while (!connected() &&
 190          GetNumSentClientHellos() <= QuicCryptoClientStream::kMaxClientHellos) {
 191     StartConnect();
 192     while (EncryptionBeingEstablished()) {
 193       WaitForEvents();
 194     }
 195     if (FLAGS_enable_quic_stateless_reject_support && connected() &&
 196         !data_to_resend_on_connect_.empty()) {
 197       // A connection has been established and there was previously queued data
 198       // to resend.  Resend it and empty the queue.
 199       for (QuicDataToResend* data : data_to_resend_on_connect_) {
 200         data->Resend();
 201       }
 202       STLDeleteElements(&data_to_resend_on_connect_);
 203     }
 204     if (session() != nullptr &&
 205         session()->error() != QUIC_CRYPTO_HANDSHAKE_STATELESS_REJECT) {
 206       // We've successfully created a session but we're not connected, and there
 207       // is no stateless reject to recover from.  Give up trying.
 208       break;
 209     }
 210   }
 211   if (!connected() &&
 212       GetNumSentClientHellos() > QuicCryptoClientStream::kMaxClientHellos &&
 213       session() != nullptr &&
 214       session()->error() == QUIC_CRYPTO_HANDSHAKE_STATELESS_REJECT) {
 215     // The overall connection failed due too many stateless rejects.
 216     set_connection_error(QUIC_CRYPTO_TOO_MANY_REJECTS);
 217   }
 218   return session()->connection()->connected();
 219 }
 220
 221 void QuicClient::StartConnect() {
 222   DCHECK(initialized_);
 223   DCHECK(!connected());
 224
 225   QuicPacketWriter* writer = CreateQuicPacketWriter();
 226
 227   DummyPacketWriterFactory factory(writer);
 228
 229   if (connected_or_attempting_connect()) {
 230     // Before we destroy the last session and create a new one, gather its stats
 231     // and update the stats for the overall connection.
 232     UpdateStats();
 233     if (session()->error() == QUIC_CRYPTO_HANDSHAKE_STATELESS_REJECT) {
 234       // If the last error was due to a stateless reject, queue up the data to
 235       // be resent on the next successful connection.
 236       // TODO(jokulik): I'm a little bit concerned about ordering here.  Maybe
 237       // we should just maintain one queue?
 238       DCHECK(data_to_resend_on_connect_.empty());
 239       data_to_resend_on_connect_.swap(data_sent_before_handshake_);
 240     }
 241   }
 242
 243   CreateQuicClientSession(new QuicConnection(
 244       GetNextConnectionId(), server_address_, helper_.get(), factory,
 245       /* owns_writer= */ false, Perspective::IS_CLIENT, server_id().is_https(),
 246       supported_versions()));
 247
 248   // Reset |writer_| after |session()| so that the old writer outlives the old
 249   // session.
 250   set_writer(writer);
 251   session()->Initialize();
 252   session()->CryptoConnect();
 253   set_connected_or_attempting_connect(true);
 254 }
 255
 256 void QuicClient::Disconnect() {
 257   DCHECK(initialized_);
 258
 259   if (connected()) {
 260     session()->connection()->SendConnectionClose(QUIC_PEER_GOING_AWAY);
 261   }
 262   STLDeleteElements(&data_to_resend_on_connect_);
 263   STLDeleteElements(&data_sent_before_handshake_);
 264
 265   CleanUpUDPSocket();
 266
 267   initialized_ = false;
 268 }
 269
 270 void QuicClient::CleanUpUDPSocket() {
 271   CleanUpUDPSocketImpl();
 272 }
 273
 274 void QuicClient::CleanUpUDPSocketImpl() {
 275   if (fd_ > -1) {
 276     epoll_server_->UnregisterFD(fd_);
 277     int rc = close(fd_);
 278     DCHECK_EQ(0, rc);
 279     fd_ = -1;
 280   }
 281 }
 282
 283 void QuicClient::SendRequest(const BalsaHeaders& headers,
 284                              StringPiece body,
 285                              bool fin) {
 286   QuicSpdyClientStream* stream = CreateReliableClientStream();
 287   if (stream == nullptr) {
 288     LOG(DFATAL) << "stream creation failed!";
 289     return;
 290   }
 291   stream->set_visitor(this);
 292   stream->SendRequest(
 293       SpdyBalsaUtils::RequestHeadersToSpdyHeaders(headers, stream->version()),
 294       body, fin);
 295   if (FLAGS_enable_quic_stateless_reject_support) {
 296     // Record this in case we need to resend.
 297     auto new_headers = new BalsaHeaders;
 298     new_headers->CopyFrom(headers);
 299     auto data_to_resend =
 300         new ClientQuicDataToResend(new_headers, body, fin, this);
 301     MaybeAddQuicDataToResend(data_to_resend);
 302   }
 303 }
 304
 305 void QuicClient::MaybeAddQuicDataToResend(QuicDataToResend* data_to_resend) {
 306   DCHECK(FLAGS_enable_quic_stateless_reject_support);
 307   if (session()->IsCryptoHandshakeConfirmed()) {
 308     // The handshake is confirmed.  No need to continue saving requests to
 309     // resend.
 310     STLDeleteElements(&data_sent_before_handshake_);
 311     delete data_to_resend;
 312     return;
 313   }
 314
 315   // The handshake is not confirmed.  Push the data onto the queue of data to
 316   // resend if statelessly rejected.
 317   data_sent_before_handshake_.push_back(data_to_resend);
 318 }
 319
 320 void QuicClient::SendRequestAndWaitForResponse(
 321     const BalsaHeaders& headers,
 322     StringPiece body,
 323     bool fin) {
 324   SendRequest(headers, body, fin);
 325   while (WaitForEvents()) {}
 326 }
 327
 328 void QuicClient::SendRequestsAndWaitForResponse(
 329     const vector<string>& url_list) {
 330   for (size_t i = 0; i < url_list.size(); ++i) {
 331     BalsaHeaders headers;
 332     headers.SetRequestFirstlineFromStringPieces("GET", url_list[i], "HTTP/1.1");
 333     SendRequest(headers, "", true);
 334   }
 335   while (WaitForEvents()) {}
 336 }
 337
 338 bool QuicClient::WaitForEvents() {
 339   DCHECK(connected());
 340
 341   epoll_server_->WaitForEventsAndExecuteCallbacks();
 342
 343   DCHECK(session() != nullptr);
 344   if (!connected() &&
 345       session()->error() == QUIC_CRYPTO_HANDSHAKE_STATELESS_REJECT) {
 346     DCHECK(FLAGS_enable_quic_stateless_reject_support);
 347     DVLOG(1) << "Detected stateless reject while waiting for events.  "
 348              << "Attempting to reconnect.";
 349     Connect();
 350   }
 351
 352   return session()->num_active_requests() != 0;
 353 }
 354
 355 bool QuicClient::MigrateSocket(const IPAddressNumber& new_host) {
 356   if (!connected()) {
 357     return false;
 358   }
 359
 360   CleanUpUDPSocket();
 361
 362   bind_to_address_ = new_host;
 363   if (!CreateUDPSocket()) {
 364     return false;
 365   }
 366
 367   epoll_server_->RegisterFD(fd_, this, kEpollFlags);
 368   session()->connection()->SetSelfAddress(client_address_);
 369
 370   QuicPacketWriter* writer = CreateQuicPacketWriter();
 371   DummyPacketWriterFactory factory(writer);
 372   set_writer(writer);
 373   session()->connection()->SetQuicPacketWriter(writer, false);
 374
 375   return true;
 376 }
 377
 378 void QuicClient::OnEvent(int fd, EpollEvent* event) {
 379   DCHECK_EQ(fd, fd_);
 380
 381   if (event->in_events & EPOLLIN) {
 382     while (connected() && ReadAndProcessPacket()) {
 383     }
 384   }
 385   if (connected() && (event->in_events & EPOLLOUT)) {
 386     writer()->SetWritable();
 387     session()->connection()->OnCanWrite();
 388   }
 389   if (event->in_events & EPOLLERR) {
 390     DVLOG(1) << "Epollerr";
 391   }
 392 }
 393
 394 void QuicClient::OnClose(QuicDataStream* stream) {
 395   DCHECK(stream != nullptr);
 396   QuicSpdyClientStream* client_stream =
 397       static_cast<QuicSpdyClientStream*>(stream);
 398   BalsaHeaders headers;
 399   SpdyBalsaUtils::SpdyHeadersToResponseHeaders(client_stream->headers(),
 400                                                &headers, stream->version());
 401
 402   if (response_listener_.get() != nullptr) {
 403     response_listener_->OnCompleteResponse(
 404         stream->id(), headers, client_stream->data());
 405   }
 406
 407   // Store response headers and body.
 408   if (store_response_) {
 409     latest_response_code_ = headers.parsed_response_code();
 410     headers.DumpHeadersToString(&latest_response_headers_);
 411     latest_response_body_ = client_stream->data();
 412   }
 413 }
 414
 415 size_t QuicClient::latest_response_code() const {
 416   LOG_IF(DFATAL, !store_response_) << "Response not stored!";
 417   return latest_response_code_;
 418 }
 419
 420 const string& QuicClient::latest_response_headers() const {
 421   LOG_IF(DFATAL, !store_response_) << "Response not stored!";
 422   return latest_response_headers_;
 423 }
 424
 425 const string& QuicClient::latest_response_body() const {
 426   LOG_IF(DFATAL, !store_response_) << "Response not stored!";
 427   return latest_response_body_;
 428 }
 429
 430 QuicEpollConnectionHelper* QuicClient::CreateQuicConnectionHelper() {
 431   return new QuicEpollConnectionHelper(epoll_server_);
 432 }
 433
 434 QuicPacketWriter* QuicClient::CreateQuicPacketWriter() {
 435   return new QuicDefaultPacketWriter(fd_);
 436 }
 437
 438 int QuicClient::ReadPacket(char* buffer,
 439                            int buffer_len,
 440                            IPEndPoint* server_address,
 441                            IPAddressNumber* client_ip) {
 442   return QuicSocketUtils::ReadPacket(
 443       fd_, buffer, buffer_len,
 444       overflow_supported_ ? &packets_dropped_ : nullptr, client_ip,
 445       server_address);
 446 }
 447
 448 bool QuicClient::ReadAndProcessPacket() {
 449   // Allocate some extra space so we can send an error if the server goes over
 450   // the limit.
 451   char buf[2 * kMaxPacketSize];
 452
 453   IPEndPoint server_address;
 454   IPAddressNumber client_ip;
 455
 456   int bytes_read = ReadPacket(buf, arraysize(buf), &server_address, &client_ip);
 457
 458   if (bytes_read < 0) {
 459     return false;
 460   }
 461
 462   QuicEncryptedPacket packet(buf, bytes_read, false);
 463
 464   IPEndPoint client_address(client_ip, client_address_.port());
 465   session()->connection()->ProcessUdpPacket(client_address, server_address,
 466                                             packet);
 467   return true;
 468 }
 469
 470 }  // namespace tools
 471 }  // namespace net