Home | History | Annotate | Download | only in quic 
      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_stream_factory.h"
     19 #include "net/ssl/ssl_info.h"
     20 #include "net/udp/datagram_client_socket.h"
     21 
     22 namespace net {
     23 
     24 namespace {
     25 
     26 // Note: these values must be kept in sync with the corresponding values in:
     27 // tools/metrics/histograms/histograms.xml
     28 enum HandshakeState {
     29   STATE_STARTED = 0,
     30   STATE_ENCRYPTION_ESTABLISHED = 1,
     31   STATE_HANDSHAKE_CONFIRMED = 2,
     32   STATE_FAILED = 3,
     33   NUM_HANDSHAKE_STATES = 4
     34 };
     35 
     36 void RecordHandshakeState(HandshakeState state) {
     37   UMA_HISTOGRAM_ENUMERATION("Net.QuicHandshakeState", state,
     38                             NUM_HANDSHAKE_STATES);
     39 }
     40 
     41 }  // namespace
     42 
     43 QuicClientSession::StreamRequest::StreamRequest() : stream_(NULL) {}
     44 
     45 QuicClientSession::StreamRequest::~StreamRequest() {
     46   CancelRequest();
     47 }
     48 
     49 int QuicClientSession::StreamRequest::StartRequest(
     50     const base::WeakPtr<QuicClientSession> session,
     51     QuicReliableClientStream** stream,
     52     const CompletionCallback& callback) {
     53   session_ = session;
     54   stream_ = stream;
     55   int rv = session_->TryCreateStream(this, stream_);
     56   if (rv == ERR_IO_PENDING) {
     57     callback_ = callback;
     58   }
     59 
     60   return rv;
     61 }
     62 
     63 void QuicClientSession::StreamRequest::CancelRequest() {
     64   if (session_)
     65     session_->CancelRequest(this);
     66   session_.reset();
     67   callback_.Reset();
     68 }
     69 
     70 void QuicClientSession::StreamRequest::OnRequestCompleteSuccess(
     71     QuicReliableClientStream* stream) {
     72   session_.reset();
     73   *stream_ = stream;
     74   ResetAndReturn(&callback_).Run(OK);
     75 }
     76 
     77 void QuicClientSession::StreamRequest::OnRequestCompleteFailure(int rv) {
     78   session_.reset();
     79   ResetAndReturn(&callback_).Run(rv);
     80 }
     81 
     82 QuicClientSession::QuicClientSession(
     83     QuicConnection* connection,
     84     DatagramClientSocket* socket,
     85     QuicStreamFactory* stream_factory,
     86     QuicCryptoClientStreamFactory* crypto_client_stream_factory,
     87     const string& server_hostname,
     88     const QuicConfig& config,
     89     QuicCryptoClientConfig* crypto_config,
     90     NetLog* net_log)
     91     : QuicSession(connection, config, false),
     92       weak_factory_(this),
     93       stream_factory_(stream_factory),
     94       socket_(socket),
     95       read_buffer_(new IOBufferWithSize(kMaxPacketSize)),
     96       read_pending_(false),
     97       num_total_streams_(0),
     98       net_log_(BoundNetLog::Make(net_log, NetLog::SOURCE_QUIC_SESSION)),
     99       logger_(net_log_) {
    100   crypto_stream_.reset(
    101       crypto_client_stream_factory ?
    102           crypto_client_stream_factory->CreateQuicCryptoClientStream(
    103               server_hostname, this, crypto_config) :
    104           new QuicCryptoClientStream(server_hostname, this, crypto_config));
    105 
    106   connection->set_debug_visitor(&logger_);
    107   // TODO(rch): pass in full host port proxy pair
    108   net_log_.BeginEvent(
    109       NetLog::TYPE_QUIC_SESSION,
    110       NetLog::StringCallback("host", &server_hostname));
    111 }
    112 
    113 QuicClientSession::~QuicClientSession() {
    114   connection()->set_debug_visitor(NULL);
    115   net_log_.EndEvent(NetLog::TYPE_QUIC_SESSION);
    116 
    117   while (!stream_requests_.empty()) {
    118     StreamRequest* request = stream_requests_.front();
    119     stream_requests_.pop_front();
    120     request->OnRequestCompleteFailure(ERR_ABORTED);
    121   }
    122 
    123   if (IsEncryptionEstablished())
    124     RecordHandshakeState(STATE_ENCRYPTION_ESTABLISHED);
    125   if (IsCryptoHandshakeConfirmed())
    126     RecordHandshakeState(STATE_HANDSHAKE_CONFIRMED);
    127   else
    128     RecordHandshakeState(STATE_FAILED);
    129 
    130   UMA_HISTOGRAM_COUNTS("Net.QuicNumSentClientHellos",
    131                        crypto_stream_->num_sent_client_hellos());
    132   if (IsCryptoHandshakeConfirmed()) {
    133     UMA_HISTOGRAM_COUNTS("Net.QuicNumSentClientHellosCryptoHandshakeConfirmed",
    134                          crypto_stream_->num_sent_client_hellos());
    135   }
    136 }
    137 
    138 int QuicClientSession::TryCreateStream(StreamRequest* request,
    139                                        QuicReliableClientStream** stream) {
    140   if (!crypto_stream_->encryption_established()) {
    141     DLOG(DFATAL) << "Encryption not established.";
    142     return ERR_CONNECTION_CLOSED;
    143   }
    144 
    145   if (goaway_received()) {
    146     DLOG(INFO) << "Going away.";
    147     return ERR_CONNECTION_CLOSED;
    148   }
    149 
    150   if (!connection()->connected()) {
    151     DLOG(INFO) << "Already closed.";
    152     return ERR_CONNECTION_CLOSED;
    153   }
    154 
    155   if (GetNumOpenStreams() < get_max_open_streams()) {
    156     *stream = CreateOutgoingReliableStreamImpl();
    157     return OK;
    158   }
    159 
    160   stream_requests_.push_back(request);
    161   return ERR_IO_PENDING;
    162 }
    163 
    164 void QuicClientSession::CancelRequest(StreamRequest* request) {
    165   // Remove |request| from the queue while preserving the order of the
    166   // other elements.
    167   StreamRequestQueue::iterator it =
    168       std::find(stream_requests_.begin(), stream_requests_.end(), request);
    169   if (it != stream_requests_.end()) {
    170     it = stream_requests_.erase(it);
    171   }
    172 }
    173 
    174 QuicReliableClientStream* QuicClientSession::CreateOutgoingReliableStream() {
    175   if (!crypto_stream_->encryption_established()) {
    176     DLOG(INFO) << "Encryption not active so no outgoing stream created.";
    177     return NULL;
    178   }
    179   if (GetNumOpenStreams() >= get_max_open_streams()) {
    180     DLOG(INFO) << "Failed to create a new outgoing stream. "
    181                << "Already " << GetNumOpenStreams() << " open.";
    182     return NULL;
    183   }
    184   if (goaway_received()) {
    185     DLOG(INFO) << "Failed to create a new outgoing stream. "
    186                << "Already received goaway.";
    187     return NULL;
    188   }
    189 
    190   return CreateOutgoingReliableStreamImpl();
    191 }
    192 
    193 QuicReliableClientStream*
    194 QuicClientSession::CreateOutgoingReliableStreamImpl() {
    195   DCHECK(connection()->connected());
    196   QuicReliableClientStream* stream =
    197       new QuicReliableClientStream(GetNextStreamId(), this, net_log_);
    198   ActivateStream(stream);
    199   ++num_total_streams_;
    200   return stream;
    201 }
    202 
    203 QuicCryptoClientStream* QuicClientSession::GetCryptoStream() {
    204   return crypto_stream_.get();
    205 };
    206 
    207 bool QuicClientSession::GetSSLInfo(SSLInfo* ssl_info) {
    208   DCHECK(crypto_stream_.get());
    209   return crypto_stream_->GetSSLInfo(ssl_info);
    210 }
    211 
    212 int QuicClientSession::CryptoConnect(const CompletionCallback& callback) {
    213   RecordHandshakeState(STATE_STARTED);
    214   if (!crypto_stream_->CryptoConnect()) {
    215     // TODO(wtc): change crypto_stream_.CryptoConnect() to return a
    216     // QuicErrorCode and map it to a net error code.
    217     return ERR_CONNECTION_FAILED;
    218   }
    219 
    220   if (IsEncryptionEstablished()) {
    221     return OK;
    222   }
    223 
    224   callback_ = callback;
    225   return ERR_IO_PENDING;
    226 }
    227 
    228 ReliableQuicStream* QuicClientSession::CreateIncomingReliableStream(
    229     QuicStreamId id) {
    230   DLOG(ERROR) << "Server push not supported";
    231   return NULL;
    232 }
    233 
    234 void QuicClientSession::CloseStream(QuicStreamId stream_id) {
    235   QuicSession::CloseStream(stream_id);
    236 
    237   if (GetNumOpenStreams() < get_max_open_streams() &&
    238       !stream_requests_.empty() &&
    239       crypto_stream_->encryption_established() &&
    240       !goaway_received() &&
    241       connection()->connected()) {
    242     StreamRequest* request = stream_requests_.front();
    243     stream_requests_.pop_front();
    244     request->OnRequestCompleteSuccess(CreateOutgoingReliableStreamImpl());
    245   }
    246 
    247   if (GetNumOpenStreams() == 0) {
    248     stream_factory_->OnIdleSession(this);
    249   }
    250 }
    251 
    252 void QuicClientSession::OnCryptoHandshakeEvent(CryptoHandshakeEvent event) {
    253   if (!callback_.is_null()) {
    254     // TODO(rtenneti): Currently for all CryptoHandshakeEvent events, callback_
    255     // could be called because there are no error events in CryptoHandshakeEvent
    256     // enum. If error events are added to CryptoHandshakeEvent, then the
    257     // following code needs to changed.
    258     base::ResetAndReturn(&callback_).Run(OK);
    259   }
    260   QuicSession::OnCryptoHandshakeEvent(event);
    261 }
    262 
    263 void QuicClientSession::ConnectionClose(QuicErrorCode error, bool from_peer) {
    264   UMA_HISTOGRAM_SPARSE_SLOWLY("Net.QuicSession.ConnectionCloseErrorCode",
    265                               error);
    266   UMA_HISTOGRAM_SPARSE_SLOWLY("Net.QuicSession.QuicVersion",
    267                               connection()->version());
    268   if (!callback_.is_null()) {
    269     base::ResetAndReturn(&callback_).Run(ERR_QUIC_PROTOCOL_ERROR);
    270   }
    271   QuicSession::ConnectionClose(error, from_peer);
    272   NotifyFactoryOfSessionCloseLater();
    273 }
    274 
    275 void QuicClientSession::StartReading() {
    276   if (read_pending_) {
    277     return;
    278   }
    279   read_pending_ = true;
    280   int rv = socket_->Read(read_buffer_.get(),
    281                          read_buffer_->size(),
    282                          base::Bind(&QuicClientSession::OnReadComplete,
    283                                     weak_factory_.GetWeakPtr()));
    284   if (rv == ERR_IO_PENDING) {
    285     return;
    286   }
    287 
    288   // Data was read, process it.
    289   // Schedule the work through the message loop to avoid recursive
    290   // callbacks.
    291   base::MessageLoop::current()->PostTask(
    292       FROM_HERE,
    293       base::Bind(&QuicClientSession::OnReadComplete,
    294                  weak_factory_.GetWeakPtr(), rv));
    295 }
    296 
    297 void QuicClientSession::CloseSessionOnError(int error) {
    298   UMA_HISTOGRAM_SPARSE_SLOWLY("Net.QuicSession.CloseSessionOnError", -error);
    299   CloseSessionOnErrorInner(error);
    300   NotifyFactoryOfSessionClose();
    301 }
    302 
    303 void QuicClientSession::CloseSessionOnErrorInner(int error) {
    304   if (!callback_.is_null()) {
    305     base::ResetAndReturn(&callback_).Run(error);
    306   }
    307   while (!streams()->empty()) {
    308     ReliableQuicStream* stream = streams()->begin()->second;
    309     QuicStreamId id = stream->id();
    310     static_cast<QuicReliableClientStream*>(stream)->OnError(error);
    311     CloseStream(id);
    312   }
    313   net_log_.AddEvent(
    314       NetLog::TYPE_QUIC_SESSION_CLOSE_ON_ERROR,
    315       NetLog::IntegerCallback("net_error", error));
    316 
    317   connection()->CloseConnection(QUIC_INTERNAL_ERROR, false);
    318   DCHECK(!connection()->connected());
    319 }
    320 
    321 base::Value* QuicClientSession::GetInfoAsValue(const HostPortPair& pair) const {
    322   base::DictionaryValue* dict = new base::DictionaryValue();
    323   dict->SetString("host_port_pair", pair.ToString());
    324   dict->SetString("version", QuicVersionToString(connection()->version()));
    325   dict->SetInteger("open_streams", GetNumOpenStreams());
    326   dict->SetInteger("total_streams", num_total_streams_);
    327   dict->SetString("peer_address", peer_address().ToString());
    328   dict->SetString("guid", base::Uint64ToString(guid()));
    329   return dict;
    330 }
    331 
    332 base::WeakPtr<QuicClientSession> QuicClientSession::GetWeakPtr() {
    333   return weak_factory_.GetWeakPtr();
    334 }
    335 
    336 void QuicClientSession::OnReadComplete(int result) {
    337   read_pending_ = false;
    338   if (result == 0)
    339     result = ERR_CONNECTION_CLOSED;
    340 
    341   if (result < 0) {
    342     DLOG(INFO) << "Closing session on read error: " << result;
    343     CloseSessionOnErrorInner(result);
    344     NotifyFactoryOfSessionCloseLater();
    345     return;
    346   }
    347 
    348   scoped_refptr<IOBufferWithSize> buffer(read_buffer_);
    349   read_buffer_ = new IOBufferWithSize(kMaxPacketSize);
    350   QuicEncryptedPacket packet(buffer->data(), result);
    351   IPEndPoint local_address;
    352   IPEndPoint peer_address;
    353   socket_->GetLocalAddress(&local_address);
    354   socket_->GetPeerAddress(&peer_address);
    355   // ProcessUdpPacket might result in |this| being deleted, so we
    356   // use a weak pointer to be safe.
    357   connection()->ProcessUdpPacket(local_address, peer_address, packet);
    358   if (!connection()->connected()) {
    359     stream_factory_->OnSessionClose(this);
    360     return;
    361   }
    362   StartReading();
    363 }
    364 
    365 void QuicClientSession::NotifyFactoryOfSessionCloseLater() {
    366   DCHECK_EQ(0u, GetNumOpenStreams());
    367   DCHECK(!connection()->connected());
    368   base::MessageLoop::current()->PostTask(
    369       FROM_HERE,
    370       base::Bind(&QuicClientSession::NotifyFactoryOfSessionClose,
    371                  weak_factory_.GetWeakPtr()));
    372 }
    373 
    374 void QuicClientSession::NotifyFactoryOfSessionClose() {
    375   DCHECK_EQ(0u, GetNumOpenStreams());
    376   DCHECK(stream_factory_);
    377   // Will delete |this|.
    378   stream_factory_->OnSessionClose(this);
    379 }
    380 
    381 }  // namespace net
    382