src/net/quic/quic_client_session.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/quic/quic_client_session.h"
   6
   7 #include "base/callback_helpers.h"
   8 #include "base/message_loop/message_loop.h"
   9 #include "base/metrics/histogram.h"
  10 #include "base/metrics/sparse_histogram.h"
  11 #include "base/stl_util.h"
  12 #include "base/strings/string_number_conversions.h"
  13 #include "base/values.h"
  14 #include "net/base/io_buffer.h"
  15 #include "net/base/net_errors.h"
  16 #include "net/quic/quic_connection_helper.h"
  17 #include "net/quic/quic_crypto_client_stream_factory.h"
  18 #include "net/quic/quic_default_packet_writer.h"
  19 #include "net/quic/quic_stream_factory.h"
  20 #include "net/ssl/ssl_info.h"
  21 #include "net/udp/datagram_client_socket.h"
  22
  23 namespace net {
  24
  25 namespace {
  26
  27 // Note: these values must be kept in sync with the corresponding values in:
  28 // tools/metrics/histograms/histograms.xml
  29 enum HandshakeState {
  30   STATE_STARTED = 0,
  31   STATE_ENCRYPTION_ESTABLISHED = 1,
  32   STATE_HANDSHAKE_CONFIRMED = 2,
  33   STATE_FAILED = 3,
  34   NUM_HANDSHAKE_STATES = 4
  35 };
  36
  37 void RecordHandshakeState(HandshakeState state) {
  38   UMA_HISTOGRAM_ENUMERATION("Net.QuicHandshakeState", state,
  39                             NUM_HANDSHAKE_STATES);
  40 }
  41
  42 }  // namespace
  43
  44 QuicClientSession::StreamRequest::StreamRequest() : stream_(NULL) {}
  45
  46 QuicClientSession::StreamRequest::~StreamRequest() {
  47   CancelRequest();
  48 }
  49
  50 int QuicClientSession::StreamRequest::StartRequest(
  51     const base::WeakPtr<QuicClientSession>& session,
  52     QuicReliableClientStream** stream,
  53     const CompletionCallback& callback) {
  54   session_ = session;
  55   stream_ = stream;
  56   int rv = session_->TryCreateStream(this, stream_);
  57   if (rv == ERR_IO_PENDING) {
  58     callback_ = callback;
  59   }
  60
  61   return rv;
  62 }
  63
  64 void QuicClientSession::StreamRequest::CancelRequest() {
  65   if (session_)
  66     session_->CancelRequest(this);
  67   session_.reset();
  68   callback_.Reset();
  69 }
  70
  71 void QuicClientSession::StreamRequest::OnRequestCompleteSuccess(
  72     QuicReliableClientStream* stream) {
  73   session_.reset();
  74   *stream_ = stream;
  75   ResetAndReturn(&callback_).Run(OK);
  76 }
  77
  78 void QuicClientSession::StreamRequest::OnRequestCompleteFailure(int rv) {
  79   session_.reset();
  80   ResetAndReturn(&callback_).Run(rv);
  81 }
  82
  83 QuicClientSession::QuicClientSession(
  84     QuicConnection* connection,
  85     scoped_ptr<DatagramClientSocket> socket,
  86     scoped_ptr<QuicDefaultPacketWriter> writer,
  87     QuicStreamFactory* stream_factory,
  88     QuicCryptoClientStreamFactory* crypto_client_stream_factory,
  89     const string& server_hostname,
  90     const QuicConfig& config,
  91     QuicCryptoClientConfig* crypto_config,
  92     NetLog* net_log)
  93     : QuicSession(connection, config, false),
  94       require_confirmation_(false),
  95       stream_factory_(stream_factory),
  96       socket_(socket.Pass()),
  97       writer_(writer.Pass()),
  98       read_buffer_(new IOBufferWithSize(kMaxPacketSize)),
  99       read_pending_(false),
 100       num_total_streams_(0),
 101       net_log_(BoundNetLog::Make(net_log, NetLog::SOURCE_QUIC_SESSION)),
 102       logger_(net_log_),
 103       num_packets_read_(0),
 104       weak_factory_(this) {
 105   crypto_stream_.reset(
 106       crypto_client_stream_factory ?
 107           crypto_client_stream_factory->CreateQuicCryptoClientStream(
 108               server_hostname, this, crypto_config) :
 109           new QuicCryptoClientStream(server_hostname, this, crypto_config));
 110
 111   connection->set_debug_visitor(&logger_);
 112   // TODO(rch): pass in full host port proxy pair
 113   net_log_.BeginEvent(
 114       NetLog::TYPE_QUIC_SESSION,
 115       NetLog::StringCallback("host", &server_hostname));
 116 }
 117
 118 QuicClientSession::~QuicClientSession() {
 119   // The session must be closed before it is destroyed.
 120   DCHECK(streams()->empty());
 121   CloseAllStreams(ERR_UNEXPECTED);
 122   DCHECK(observers_.empty());
 123   CloseAllObservers(ERR_UNEXPECTED);
 124
 125   connection()->set_debug_visitor(NULL);
 126   net_log_.EndEvent(NetLog::TYPE_QUIC_SESSION);
 127
 128   while (!stream_requests_.empty()) {
 129     StreamRequest* request = stream_requests_.front();
 130     stream_requests_.pop_front();
 131     request->OnRequestCompleteFailure(ERR_ABORTED);
 132   }
 133
 134   if (IsEncryptionEstablished())
 135     RecordHandshakeState(STATE_ENCRYPTION_ESTABLISHED);
 136   if (IsCryptoHandshakeConfirmed())
 137     RecordHandshakeState(STATE_HANDSHAKE_CONFIRMED);
 138   else
 139     RecordHandshakeState(STATE_FAILED);
 140
 141   UMA_HISTOGRAM_COUNTS("Net.QuicNumSentClientHellos",
 142                        crypto_stream_->num_sent_client_hellos());
 143   if (IsCryptoHandshakeConfirmed()) {
 144     UMA_HISTOGRAM_COUNTS("Net.QuicNumSentClientHellosCryptoHandshakeConfirmed",
 145                          crypto_stream_->num_sent_client_hellos());
 146   }
 147
 148   UMA_HISTOGRAM_COUNTS("Net.QuicSession.NumTotalStreams", num_total_streams_);
 149 }
 150
 151 bool QuicClientSession::OnStreamFrames(
 152     const std::vector<QuicStreamFrame>& frames) {
 153   // Record total number of stream frames.
 154   UMA_HISTOGRAM_COUNTS("Net.QuicNumStreamFramesInPacket", frames.size());
 155
 156   // Record number of frames per stream in packet.
 157   typedef std::map<QuicStreamId, size_t> FrameCounter;
 158   FrameCounter frames_per_stream;
 159   for (size_t i = 0; i < frames.size(); ++i) {
 160     frames_per_stream[frames[i].stream_id]++;
 161   }
 162   for (FrameCounter::const_iterator it = frames_per_stream.begin();
 163        it != frames_per_stream.end(); ++it) {
 164     UMA_HISTOGRAM_COUNTS("Net.QuicNumStreamFramesPerStreamInPacket",
 165                          it->second);
 166   }
 167
 168   return QuicSession::OnStreamFrames(frames);
 169 }
 170
 171 void QuicClientSession::AddObserver(Observer* observer) {
 172   DCHECK(!ContainsKey(observers_, observer));
 173   observers_.insert(observer);
 174 }
 175
 176 void QuicClientSession::RemoveObserver(Observer* observer) {
 177   DCHECK(ContainsKey(observers_, observer));
 178   observers_.erase(observer);
 179 }
 180
 181 int QuicClientSession::TryCreateStream(StreamRequest* request,
 182                                        QuicReliableClientStream** stream) {
 183   if (!crypto_stream_->encryption_established()) {
 184     DLOG(DFATAL) << "Encryption not established.";
 185     return ERR_CONNECTION_CLOSED;
 186   }
 187
 188   if (goaway_received()) {
 189     DLOG(INFO) << "Going away.";
 190     return ERR_CONNECTION_CLOSED;
 191   }
 192
 193   if (!connection()->connected()) {
 194     DLOG(INFO) << "Already closed.";
 195     return ERR_CONNECTION_CLOSED;
 196   }
 197
 198   if (GetNumOpenStreams() < get_max_open_streams()) {
 199     *stream = CreateOutgoingReliableStreamImpl();
 200     return OK;
 201   }
 202
 203   stream_requests_.push_back(request);
 204   return ERR_IO_PENDING;
 205 }
 206
 207 void QuicClientSession::CancelRequest(StreamRequest* request) {
 208   // Remove |request| from the queue while preserving the order of the
 209   // other elements.
 210   StreamRequestQueue::iterator it =
 211       std::find(stream_requests_.begin(), stream_requests_.end(), request);
 212   if (it != stream_requests_.end()) {
 213     it = stream_requests_.erase(it);
 214   }
 215 }
 216
 217 QuicReliableClientStream* QuicClientSession::CreateOutgoingReliableStream() {
 218   if (!crypto_stream_->encryption_established()) {
 219     DLOG(INFO) << "Encryption not active so no outgoing stream created.";
 220     return NULL;
 221   }
 222   if (GetNumOpenStreams() >= get_max_open_streams()) {
 223     DLOG(INFO) << "Failed to create a new outgoing stream. "
 224                << "Already " << GetNumOpenStreams() << " open.";
 225     return NULL;
 226   }
 227   if (goaway_received()) {
 228     DLOG(INFO) << "Failed to create a new outgoing stream. "
 229                << "Already received goaway.";
 230     return NULL;
 231   }
 232
 233   return CreateOutgoingReliableStreamImpl();
 234 }
 235
 236 QuicReliableClientStream*
 237 QuicClientSession::CreateOutgoingReliableStreamImpl() {
 238   DCHECK(connection()->connected());
 239   QuicReliableClientStream* stream =
 240       new QuicReliableClientStream(GetNextStreamId(), this, net_log_);
 241   ActivateStream(stream);
 242   ++num_total_streams_;
 243   UMA_HISTOGRAM_COUNTS("Net.QuicSession.NumOpenStreams", GetNumOpenStreams());
 244   return stream;
 245 }
 246
 247 QuicCryptoClientStream* QuicClientSession::GetCryptoStream() {
 248   return crypto_stream_.get();
 249 };
 250
 251 bool QuicClientSession::GetSSLInfo(SSLInfo* ssl_info) {
 252   DCHECK(crypto_stream_.get());
 253   return crypto_stream_->GetSSLInfo(ssl_info);
 254 }
 255
 256 int QuicClientSession::CryptoConnect(bool require_confirmation,
 257                                      const CompletionCallback& callback) {
 258   require_confirmation_ = require_confirmation;
 259   RecordHandshakeState(STATE_STARTED);
 260   if (!crypto_stream_->CryptoConnect()) {
 261     // TODO(wtc): change crypto_stream_.CryptoConnect() to return a
 262     // QuicErrorCode and map it to a net error code.
 263     return ERR_CONNECTION_FAILED;
 264   }
 265
 266   bool can_notify = require_confirmation_ ?
 267       IsCryptoHandshakeConfirmed() : IsEncryptionEstablished();
 268   if (can_notify) {
 269     return OK;
 270   }
 271
 272   callback_ = callback;
 273   return ERR_IO_PENDING;
 274 }
 275
 276 int QuicClientSession::GetNumSentClientHellos() const {
 277   return crypto_stream_->num_sent_client_hellos();
 278 }
 279
 280 ReliableQuicStream* QuicClientSession::CreateIncomingReliableStream(
 281     QuicStreamId id) {
 282   DLOG(ERROR) << "Server push not supported";
 283   return NULL;
 284 }
 285
 286 void QuicClientSession::CloseStream(QuicStreamId stream_id) {
 287   QuicSession::CloseStream(stream_id);
 288   OnClosedStream();
 289 }
 290
 291 void QuicClientSession::SendRstStream(QuicStreamId id,
 292                                       QuicRstStreamErrorCode error) {
 293   QuicSession::SendRstStream(id, error);
 294   OnClosedStream();
 295 }
 296
 297 void QuicClientSession::OnClosedStream() {
 298   if (GetNumOpenStreams() < get_max_open_streams() &&
 299       !stream_requests_.empty() &&
 300       crypto_stream_->encryption_established() &&
 301       !goaway_received() &&
 302       connection()->connected()) {
 303     StreamRequest* request = stream_requests_.front();
 304     stream_requests_.pop_front();
 305     request->OnRequestCompleteSuccess(CreateOutgoingReliableStreamImpl());
 306   }
 307
 308   if (GetNumOpenStreams() == 0) {
 309     stream_factory_->OnIdleSession(this);
 310   }
 311 }
 312
 313 void QuicClientSession::OnCryptoHandshakeEvent(CryptoHandshakeEvent event) {
 314   if (!callback_.is_null() &&
 315       (!require_confirmation_ || event == HANDSHAKE_CONFIRMED)) {
 316     // TODO(rtenneti): Currently for all CryptoHandshakeEvent events, callback_
 317     // could be called because there are no error events in CryptoHandshakeEvent
 318     // enum. If error events are added to CryptoHandshakeEvent, then the
 319     // following code needs to changed.
 320     base::ResetAndReturn(&callback_).Run(OK);
 321   }
 322   if (event == HANDSHAKE_CONFIRMED) {
 323     ObserverSet::iterator it = observers_.begin();
 324     while (it != observers_.end()) {
 325       Observer* observer = *it;
 326       ++it;
 327       observer->OnCryptoHandshakeConfirmed();
 328     }
 329   }
 330   QuicSession::OnCryptoHandshakeEvent(event);
 331 }
 332
 333 void QuicClientSession::OnCryptoHandshakeMessageSent(
 334     const CryptoHandshakeMessage& message) {
 335   logger_.OnCryptoHandshakeMessageSent(message);
 336 }
 337
 338 void QuicClientSession::OnCryptoHandshakeMessageReceived(
 339     const CryptoHandshakeMessage& message) {
 340   logger_.OnCryptoHandshakeMessageReceived(message);
 341 }
 342
 343 void QuicClientSession::OnConnectionClosed(QuicErrorCode error,
 344                                            bool from_peer) {
 345   DCHECK(!connection()->connected());
 346   logger_.OnConnectionClosed(error, from_peer);
 347   if (from_peer) {
 348     UMA_HISTOGRAM_SPARSE_SLOWLY(
 349         "Net.QuicSession.ConnectionCloseErrorCodeServer", error);
 350   } else {
 351     UMA_HISTOGRAM_SPARSE_SLOWLY(
 352         "Net.QuicSession.ConnectionCloseErrorCodeClient", error);
 353   }
 354
 355   if (error == QUIC_CONNECTION_TIMED_OUT) {
 356     UMA_HISTOGRAM_COUNTS(
 357         "Net.QuicSession.ConnectionClose.NumOpenStreams.TimedOut",
 358         GetNumOpenStreams());
 359     if (!IsCryptoHandshakeConfirmed()) {
 360       // If there have been any streams created, they were 0-RTT speculative
 361       // requests that have not be serviced.
 362       UMA_HISTOGRAM_COUNTS(
 363           "Net.QuicSession.ConnectionClose.NumTotalStreams.HandshakeTimedOut",
 364           num_total_streams_);
 365     }
 366   }
 367
 368   UMA_HISTOGRAM_SPARSE_SLOWLY("Net.QuicSession.QuicVersion",
 369                               connection()->version());
 370   NotifyFactoryOfSessionGoingAway();
 371   if (!callback_.is_null()) {
 372     base::ResetAndReturn(&callback_).Run(ERR_QUIC_PROTOCOL_ERROR);
 373   }
 374   socket_->Close();
 375   QuicSession::OnConnectionClosed(error, from_peer);
 376   NotifyFactoryOfSessionClosedLater();
 377 }
 378
 379 void QuicClientSession::OnSuccessfulVersionNegotiation(
 380     const QuicVersion& version) {
 381   logger_.OnSuccessfulVersionNegotiation(version);
 382   QuicSession::OnSuccessfulVersionNegotiation(version);
 383 }
 384
 385 void QuicClientSession::StartReading() {
 386   if (read_pending_) {
 387     return;
 388   }
 389   read_pending_ = true;
 390   int rv = socket_->Read(read_buffer_.get(),
 391                          read_buffer_->size(),
 392                          base::Bind(&QuicClientSession::OnReadComplete,
 393                                     weak_factory_.GetWeakPtr()));
 394   if (rv == ERR_IO_PENDING) {
 395     num_packets_read_ = 0;
 396     return;
 397   }
 398
 399   if (++num_packets_read_ > 32) {
 400     num_packets_read_ = 0;
 401     // Data was read, process it.
 402     // Schedule the work through the message loop to avoid recursive
 403     // callbacks.
 404     base::MessageLoop::current()->PostTask(
 405         FROM_HERE,
 406         base::Bind(&QuicClientSession::OnReadComplete,
 407                    weak_factory_.GetWeakPtr(), rv));
 408   } else {
 409     OnReadComplete(rv);
 410   }
 411 }
 412
 413 void QuicClientSession::CloseSessionOnError(int error) {
 414   UMA_HISTOGRAM_SPARSE_SLOWLY("Net.QuicSession.CloseSessionOnError", -error);
 415   CloseSessionOnErrorInner(error, QUIC_INTERNAL_ERROR);
 416   NotifyFactoryOfSessionClosed();
 417 }
 418
 419 void QuicClientSession::CloseSessionOnErrorInner(int net_error,
 420                                                  QuicErrorCode quic_error) {
 421   if (!callback_.is_null()) {
 422     base::ResetAndReturn(&callback_).Run(net_error);
 423   }
 424   CloseAllStreams(net_error);
 425   CloseAllObservers(net_error);
 426   net_log_.AddEvent(
 427       NetLog::TYPE_QUIC_SESSION_CLOSE_ON_ERROR,
 428       NetLog::IntegerCallback("net_error", net_error));
 429
 430   connection()->CloseConnection(quic_error, false);
 431   DCHECK(!connection()->connected());
 432 }
 433
 434 void QuicClientSession::CloseAllStreams(int net_error) {
 435   while (!streams()->empty()) {
 436     ReliableQuicStream* stream = streams()->begin()->second;
 437     QuicStreamId id = stream->id();
 438     static_cast<QuicReliableClientStream*>(stream)->OnError(net_error);
 439     CloseStream(id);
 440   }
 441 }
 442
 443 void QuicClientSession::CloseAllObservers(int net_error) {
 444   while (!observers_.empty()) {
 445     Observer* observer = *observers_.begin();
 446     observers_.erase(observer);
 447     observer->OnSessionClosed(net_error);
 448   }
 449 }
 450
 451 base::Value* QuicClientSession::GetInfoAsValue(const HostPortPair& pair) const {
 452   base::DictionaryValue* dict = new base::DictionaryValue();
 453   dict->SetString("host_port_pair", pair.ToString());
 454   dict->SetString("version", QuicVersionToString(connection()->version()));
 455   dict->SetInteger("open_streams", GetNumOpenStreams());
 456   dict->SetInteger("total_streams", num_total_streams_);
 457   dict->SetString("peer_address", peer_address().ToString());
 458   dict->SetString("guid", base::Uint64ToString(guid()));
 459   dict->SetBoolean("connected", connection()->connected());
 460   return dict;
 461 }
 462
 463 base::WeakPtr<QuicClientSession> QuicClientSession::GetWeakPtr() {
 464   return weak_factory_.GetWeakPtr();
 465 }
 466
 467 void QuicClientSession::OnReadComplete(int result) {
 468   read_pending_ = false;
 469   if (result == 0)
 470     result = ERR_CONNECTION_CLOSED;
 471
 472   if (result < 0) {
 473     DLOG(INFO) << "Closing session on read error: " << result;
 474     UMA_HISTOGRAM_SPARSE_SLOWLY("Net.QuicSession.ReadError", -result);
 475     NotifyFactoryOfSessionGoingAway();
 476     CloseSessionOnErrorInner(result, QUIC_PACKET_READ_ERROR);
 477     NotifyFactoryOfSessionClosedLater();
 478     return;
 479   }
 480
 481   scoped_refptr<IOBufferWithSize> buffer(read_buffer_);
 482   read_buffer_ = new IOBufferWithSize(kMaxPacketSize);
 483   QuicEncryptedPacket packet(buffer->data(), result);
 484   IPEndPoint local_address;
 485   IPEndPoint peer_address;
 486   socket_->GetLocalAddress(&local_address);
 487   socket_->GetPeerAddress(&peer_address);
 488   // ProcessUdpPacket might result in |this| being deleted, so we
 489   // use a weak pointer to be safe.
 490   connection()->ProcessUdpPacket(local_address, peer_address, packet);
 491   if (!connection()->connected()) {
 492     stream_factory_->OnSessionClosed(this);
 493     return;
 494   }
 495   StartReading();
 496 }
 497
 498 void QuicClientSession::NotifyFactoryOfSessionGoingAway() {
 499   if (stream_factory_)
 500     stream_factory_->OnSessionGoingAway(this);
 501 }
 502
 503 void QuicClientSession::NotifyFactoryOfSessionClosedLater() {
 504   DCHECK_EQ(0u, GetNumOpenStreams());
 505   DCHECK(!connection()->connected());
 506   base::MessageLoop::current()->PostTask(
 507       FROM_HERE,
 508       base::Bind(&QuicClientSession::NotifyFactoryOfSessionClosed,
 509                  weak_factory_.GetWeakPtr()));
 510 }
 511
 512 void QuicClientSession::NotifyFactoryOfSessionClosed() {
 513   DCHECK_EQ(0u, GetNumOpenStreams());
 514   // Will delete |this|.
 515   if (stream_factory_)
 516     stream_factory_->OnSessionClosed(this);
 517 }
 518
 519 }  // namespace net