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_stream_factory.h"
6
7#include <set>
8
9#include "base/cpu.h"
10#include "base/message_loop/message_loop.h"
11#include "base/message_loop/message_loop_proxy.h"
12#include "base/metrics/histogram.h"
13#include "base/rand_util.h"
14#include "base/stl_util.h"
15#include "base/strings/string_util.h"
16#include "base/values.h"
17#include "net/base/net_errors.h"
18#include "net/cert/cert_verifier.h"
19#include "net/dns/host_resolver.h"
20#include "net/dns/single_request_host_resolver.h"
21#include "net/http/http_server_properties.h"
22#include "net/quic/congestion_control/tcp_receiver.h"
23#include "net/quic/crypto/proof_verifier_chromium.h"
24#include "net/quic/crypto/quic_random.h"
25#include "net/quic/crypto/quic_server_info.h"
26#include "net/quic/port_suggester.h"
27#include "net/quic/quic_client_session.h"
28#include "net/quic/quic_clock.h"
29#include "net/quic/quic_connection.h"
30#include "net/quic/quic_connection_helper.h"
31#include "net/quic/quic_crypto_client_stream_factory.h"
32#include "net/quic/quic_default_packet_writer.h"
33#include "net/quic/quic_http_stream.h"
34#include "net/quic/quic_protocol.h"
35#include "net/quic/quic_server_id.h"
36#include "net/socket/client_socket_factory.h"
37
38#if defined(OS_WIN)
39#include "base/win/windows_version.h"
40#endif
41
42using std::string;
43using std::vector;
44
45namespace net {
46
47namespace {
48
49enum CreateSessionFailure {
50  CREATION_ERROR_CONNECTING_SOCKET,
51  CREATION_ERROR_SETTING_RECEIVE_BUFFER,
52  CREATION_ERROR_SETTING_SEND_BUFFER,
53  CREATION_ERROR_MAX
54};
55
56// When a connection is idle for 30 seconds it will be closed.
57const int kIdleConnectionTimeoutSeconds = 30;
58
59// The initial receive window size for both streams and sessions.
60const int32 kInitialReceiveWindowSize = 10 * 1024 * 1024;  // 10MB
61
62// The suggested initial congestion windows for a server to use.
63// TODO: This should be tested and optimized, and even better, suggest a window
64// that corresponds to historical bandwidth and min-RTT.
65// Larger initial congestion windows can, if we don't overshoot, reduce latency
66// by avoiding the RTT needed for slow start to double (and re-double) from a
67// default of 10.
68// We match SPDY's use of 32 when secure (since we'd compete with SPDY).
69const int32 kServerSecureInitialCongestionWindow = 32;
70// Be conservative, and just use double a typical TCP  ICWND for HTTP.
71const int32 kServerInecureInitialCongestionWindow = 20;
72
73void HistogramCreateSessionFailure(enum CreateSessionFailure error) {
74  UMA_HISTOGRAM_ENUMERATION("Net.QuicSession.CreationError", error,
75                            CREATION_ERROR_MAX);
76}
77
78bool IsEcdsaSupported() {
79#if defined(OS_WIN)
80  if (base::win::GetVersion() < base::win::VERSION_VISTA)
81    return false;
82#endif
83
84  return true;
85}
86
87QuicConfig InitializeQuicConfig(bool enable_pacing,
88                                bool enable_time_based_loss_detection) {
89  QuicConfig config;
90  config.SetDefaults();
91  config.EnablePacing(enable_pacing);
92  if (enable_time_based_loss_detection)
93    config.SetLossDetectionToSend(kTIME);
94  config.set_idle_connection_state_lifetime(
95      QuicTime::Delta::FromSeconds(kIdleConnectionTimeoutSeconds),
96      QuicTime::Delta::FromSeconds(kIdleConnectionTimeoutSeconds));
97  return config;
98}
99
100}  // namespace
101
102QuicStreamFactory::IpAliasKey::IpAliasKey() {}
103
104QuicStreamFactory::IpAliasKey::IpAliasKey(IPEndPoint ip_endpoint,
105                                          bool is_https)
106    : ip_endpoint(ip_endpoint),
107      is_https(is_https) {}
108
109QuicStreamFactory::IpAliasKey::~IpAliasKey() {}
110
111bool QuicStreamFactory::IpAliasKey::operator<(
112    const QuicStreamFactory::IpAliasKey& other) const {
113  if (!(ip_endpoint == other.ip_endpoint)) {
114    return ip_endpoint < other.ip_endpoint;
115  }
116  return is_https < other.is_https;
117}
118
119bool QuicStreamFactory::IpAliasKey::operator==(
120    const QuicStreamFactory::IpAliasKey& other) const {
121  return is_https == other.is_https &&
122      ip_endpoint == other.ip_endpoint;
123};
124
125// Responsible for creating a new QUIC session to the specified server, and
126// for notifying any associated requests when complete.
127class QuicStreamFactory::Job {
128 public:
129  Job(QuicStreamFactory* factory,
130      HostResolver* host_resolver,
131      const HostPortPair& host_port_pair,
132      bool is_https,
133      bool was_alternate_protocol_recently_broken,
134      PrivacyMode privacy_mode,
135      base::StringPiece method,
136      QuicServerInfo* server_info,
137      const BoundNetLog& net_log);
138
139  // Creates a new job to handle the resumption of for connecting an
140  // existing session.
141  Job(QuicStreamFactory* factory,
142      HostResolver* host_resolver,
143      QuicClientSession* session,
144      QuicServerId server_id);
145
146  ~Job();
147
148  int Run(const CompletionCallback& callback);
149
150  int DoLoop(int rv);
151  int DoResolveHost();
152  int DoResolveHostComplete(int rv);
153  int DoLoadServerInfo();
154  int DoLoadServerInfoComplete(int rv);
155  int DoConnect();
156  int DoResumeConnect();
157  int DoConnectComplete(int rv);
158
159  void OnIOComplete(int rv);
160
161  CompletionCallback callback() {
162    return callback_;
163  }
164
165  const QuicServerId server_id() const {
166    return server_id_;
167  }
168
169 private:
170  enum IoState {
171    STATE_NONE,
172    STATE_RESOLVE_HOST,
173    STATE_RESOLVE_HOST_COMPLETE,
174    STATE_LOAD_SERVER_INFO,
175    STATE_LOAD_SERVER_INFO_COMPLETE,
176    STATE_CONNECT,
177    STATE_RESUME_CONNECT,
178    STATE_CONNECT_COMPLETE,
179  };
180  IoState io_state_;
181
182  QuicStreamFactory* factory_;
183  SingleRequestHostResolver host_resolver_;
184  QuicServerId server_id_;
185  bool is_post_;
186  bool was_alternate_protocol_recently_broken_;
187  scoped_ptr<QuicServerInfo> server_info_;
188  const BoundNetLog net_log_;
189  QuicClientSession* session_;
190  CompletionCallback callback_;
191  AddressList address_list_;
192  base::TimeTicks disk_cache_load_start_time_;
193  base::WeakPtrFactory<Job> weak_factory_;
194  DISALLOW_COPY_AND_ASSIGN(Job);
195};
196
197QuicStreamFactory::Job::Job(QuicStreamFactory* factory,
198                            HostResolver* host_resolver,
199                            const HostPortPair& host_port_pair,
200                            bool is_https,
201                            bool was_alternate_protocol_recently_broken,
202                            PrivacyMode privacy_mode,
203                            base::StringPiece method,
204                            QuicServerInfo* server_info,
205                            const BoundNetLog& net_log)
206    : io_state_(STATE_RESOLVE_HOST),
207      factory_(factory),
208      host_resolver_(host_resolver),
209      server_id_(host_port_pair, is_https, privacy_mode),
210      is_post_(method == "POST"),
211      was_alternate_protocol_recently_broken_(
212          was_alternate_protocol_recently_broken),
213      server_info_(server_info),
214      net_log_(net_log),
215      session_(NULL),
216      weak_factory_(this) {}
217
218QuicStreamFactory::Job::Job(QuicStreamFactory* factory,
219                            HostResolver* host_resolver,
220                            QuicClientSession* session,
221                            QuicServerId server_id)
222    : io_state_(STATE_RESUME_CONNECT),
223      factory_(factory),
224      host_resolver_(host_resolver),  // unused
225      server_id_(server_id),
226      is_post_(false),  // unused
227      was_alternate_protocol_recently_broken_(false),  // unused
228      net_log_(session->net_log()),  // unused
229      session_(session),
230      weak_factory_(this) {}
231
232QuicStreamFactory::Job::~Job() {
233}
234
235int QuicStreamFactory::Job::Run(const CompletionCallback& callback) {
236  int rv = DoLoop(OK);
237  if (rv == ERR_IO_PENDING)
238    callback_ = callback;
239
240  return rv > 0 ? OK : rv;
241}
242
243int QuicStreamFactory::Job::DoLoop(int rv) {
244  do {
245    IoState state = io_state_;
246    io_state_ = STATE_NONE;
247    switch (state) {
248      case STATE_RESOLVE_HOST:
249        CHECK_EQ(OK, rv);
250        rv = DoResolveHost();
251        break;
252      case STATE_RESOLVE_HOST_COMPLETE:
253        rv = DoResolveHostComplete(rv);
254        break;
255      case STATE_LOAD_SERVER_INFO:
256        CHECK_EQ(OK, rv);
257        rv = DoLoadServerInfo();
258        break;
259      case STATE_LOAD_SERVER_INFO_COMPLETE:
260        rv = DoLoadServerInfoComplete(rv);
261        break;
262      case STATE_CONNECT:
263        CHECK_EQ(OK, rv);
264        rv = DoConnect();
265        break;
266      case STATE_RESUME_CONNECT:
267        CHECK_EQ(OK, rv);
268        rv = DoResumeConnect();
269        break;
270      case STATE_CONNECT_COMPLETE:
271        rv = DoConnectComplete(rv);
272        break;
273      default:
274        NOTREACHED() << "io_state_: " << io_state_;
275        break;
276    }
277  } while (io_state_ != STATE_NONE && rv != ERR_IO_PENDING);
278  return rv;
279}
280
281void QuicStreamFactory::Job::OnIOComplete(int rv) {
282  rv = DoLoop(rv);
283
284  if (rv != ERR_IO_PENDING && !callback_.is_null()) {
285    callback_.Run(rv);
286  }
287}
288
289int QuicStreamFactory::Job::DoResolveHost() {
290  // Start loading the data now, and wait for it after we resolve the host.
291  if (server_info_) {
292    disk_cache_load_start_time_ = base::TimeTicks::Now();
293    server_info_->Start();
294  }
295
296  io_state_ = STATE_RESOLVE_HOST_COMPLETE;
297  return host_resolver_.Resolve(
298      HostResolver::RequestInfo(server_id_.host_port_pair()),
299      DEFAULT_PRIORITY,
300      &address_list_,
301      base::Bind(&QuicStreamFactory::Job::OnIOComplete,
302                 weak_factory_.GetWeakPtr()),
303      net_log_);
304}
305
306int QuicStreamFactory::Job::DoResolveHostComplete(int rv) {
307  if (rv != OK)
308    return rv;
309
310  DCHECK(!factory_->HasActiveSession(server_id_));
311
312  // Inform the factory of this resolution, which will set up
313  // a session alias, if possible.
314  if (factory_->OnResolution(server_id_, address_list_)) {
315    return OK;
316  }
317
318  io_state_ = STATE_LOAD_SERVER_INFO;
319  return OK;
320}
321
322int QuicStreamFactory::Job::DoLoadServerInfo() {
323  io_state_ = STATE_LOAD_SERVER_INFO_COMPLETE;
324
325  if (!server_info_)
326    return OK;
327
328  return server_info_->WaitForDataReady(
329      base::Bind(&QuicStreamFactory::Job::OnIOComplete,
330                 weak_factory_.GetWeakPtr()));
331}
332
333int QuicStreamFactory::Job::DoLoadServerInfoComplete(int rv) {
334  if (server_info_) {
335    UMA_HISTOGRAM_TIMES("Net.QuicServerInfo.DiskCacheReadTime",
336                        base::TimeTicks::Now() - disk_cache_load_start_time_);
337  }
338
339  if (rv != OK) {
340    server_info_.reset();
341  }
342
343  io_state_ = STATE_CONNECT;
344  return OK;
345}
346
347int QuicStreamFactory::Job::DoConnect() {
348  io_state_ = STATE_CONNECT_COMPLETE;
349
350  int rv = factory_->CreateSession(server_id_, server_info_.Pass(),
351                                   address_list_, net_log_, &session_);
352  if (rv != OK) {
353    DCHECK(rv != ERR_IO_PENDING);
354    DCHECK(!session_);
355    return rv;
356  }
357
358  session_->StartReading();
359  if (!session_->connection()->connected()) {
360    return ERR_QUIC_PROTOCOL_ERROR;
361  }
362  bool require_confirmation =
363      factory_->require_confirmation() || is_post_ ||
364      was_alternate_protocol_recently_broken_;
365  rv = session_->CryptoConnect(
366      require_confirmation,
367      base::Bind(&QuicStreamFactory::Job::OnIOComplete,
368                 base::Unretained(this)));
369  return rv;
370}
371
372int QuicStreamFactory::Job::DoResumeConnect() {
373  io_state_ = STATE_CONNECT_COMPLETE;
374
375  int rv = session_->ResumeCryptoConnect(
376      base::Bind(&QuicStreamFactory::Job::OnIOComplete,
377                 base::Unretained(this)));
378
379  return rv;
380}
381
382int QuicStreamFactory::Job::DoConnectComplete(int rv) {
383  if (rv != OK)
384    return rv;
385
386  DCHECK(!factory_->HasActiveSession(server_id_));
387  // There may well now be an active session for this IP.  If so, use the
388  // existing session instead.
389  AddressList address(session_->connection()->peer_address());
390  if (factory_->OnResolution(server_id_, address)) {
391    session_->connection()->SendConnectionClose(QUIC_CONNECTION_IP_POOLED);
392    session_ = NULL;
393    return OK;
394  }
395
396  factory_->ActivateSession(server_id_, session_);
397
398  return OK;
399}
400
401QuicStreamRequest::QuicStreamRequest(QuicStreamFactory* factory)
402    : factory_(factory) {}
403
404QuicStreamRequest::~QuicStreamRequest() {
405  if (factory_ && !callback_.is_null())
406    factory_->CancelRequest(this);
407}
408
409int QuicStreamRequest::Request(const HostPortPair& host_port_pair,
410                               bool is_https,
411                               PrivacyMode privacy_mode,
412                               base::StringPiece method,
413                               const BoundNetLog& net_log,
414                               const CompletionCallback& callback) {
415  DCHECK(!stream_);
416  DCHECK(callback_.is_null());
417  DCHECK(factory_);
418  int rv = factory_->Create(host_port_pair, is_https, privacy_mode, method,
419                            net_log, this);
420  if (rv == ERR_IO_PENDING) {
421    host_port_pair_ = host_port_pair;
422    is_https_ = is_https;
423    net_log_ = net_log;
424    callback_ = callback;
425  } else {
426    factory_ = NULL;
427  }
428  if (rv == OK)
429    DCHECK(stream_);
430  return rv;
431}
432
433void QuicStreamRequest::set_stream(scoped_ptr<QuicHttpStream> stream) {
434  DCHECK(stream);
435  stream_ = stream.Pass();
436}
437
438void QuicStreamRequest::OnRequestComplete(int rv) {
439  factory_ = NULL;
440  callback_.Run(rv);
441}
442
443scoped_ptr<QuicHttpStream> QuicStreamRequest::ReleaseStream() {
444  DCHECK(stream_);
445  return stream_.Pass();
446}
447
448QuicStreamFactory::QuicStreamFactory(
449    HostResolver* host_resolver,
450    ClientSocketFactory* client_socket_factory,
451    base::WeakPtr<HttpServerProperties> http_server_properties,
452    CertVerifier* cert_verifier,
453    TransportSecurityState* transport_security_state,
454    QuicCryptoClientStreamFactory* quic_crypto_client_stream_factory,
455    QuicRandom* random_generator,
456    QuicClock* clock,
457    size_t max_packet_length,
458    const std::string& user_agent_id,
459    const QuicVersionVector& supported_versions,
460    bool enable_port_selection,
461    bool enable_pacing,
462    bool enable_time_based_loss_detection)
463    : require_confirmation_(true),
464      host_resolver_(host_resolver),
465      client_socket_factory_(client_socket_factory),
466      http_server_properties_(http_server_properties),
467      cert_verifier_(cert_verifier),
468      quic_server_info_factory_(NULL),
469      quic_crypto_client_stream_factory_(quic_crypto_client_stream_factory),
470      random_generator_(random_generator),
471      clock_(clock),
472      max_packet_length_(max_packet_length),
473      config_(InitializeQuicConfig(enable_pacing,
474                                   enable_time_based_loss_detection)),
475      supported_versions_(supported_versions),
476      enable_port_selection_(enable_port_selection),
477      port_seed_(random_generator_->RandUint64()),
478      weak_factory_(this) {
479  crypto_config_.SetDefaults();
480  crypto_config_.set_user_agent_id(user_agent_id);
481  crypto_config_.AddCanonicalSuffix(".c.youtube.com");
482  crypto_config_.AddCanonicalSuffix(".googlevideo.com");
483  crypto_config_.SetProofVerifier(
484      new ProofVerifierChromium(cert_verifier, transport_security_state));
485  base::CPU cpu;
486  if (cpu.has_aesni() && cpu.has_avx())
487    crypto_config_.PreferAesGcm();
488  if (!IsEcdsaSupported())
489    crypto_config_.DisableEcdsa();
490}
491
492QuicStreamFactory::~QuicStreamFactory() {
493  CloseAllSessions(ERR_ABORTED);
494  while (!all_sessions_.empty()) {
495    delete all_sessions_.begin()->first;
496    all_sessions_.erase(all_sessions_.begin());
497  }
498  STLDeleteValues(&active_jobs_);
499}
500
501int QuicStreamFactory::Create(const HostPortPair& host_port_pair,
502                              bool is_https,
503                              PrivacyMode privacy_mode,
504                              base::StringPiece method,
505                              const BoundNetLog& net_log,
506                              QuicStreamRequest* request) {
507  QuicServerId server_id(host_port_pair, is_https, privacy_mode);
508  if (HasActiveSession(server_id)) {
509    request->set_stream(CreateIfSessionExists(server_id, net_log));
510    return OK;
511  }
512
513  if (HasActiveJob(server_id)) {
514    Job* job = active_jobs_[server_id];
515    active_requests_[request] = job;
516    job_requests_map_[job].insert(request);
517    return ERR_IO_PENDING;
518  }
519
520  QuicServerInfo* quic_server_info = NULL;
521  if (quic_server_info_factory_) {
522    QuicCryptoClientConfig::CachedState* cached =
523        crypto_config_.LookupOrCreate(server_id);
524    DCHECK(cached);
525    if (cached->IsEmpty()) {
526      quic_server_info = quic_server_info_factory_->GetForServer(server_id);
527    }
528  }
529  bool was_alternate_protocol_recently_broken =
530      http_server_properties_ &&
531      http_server_properties_->WasAlternateProtocolRecentlyBroken(
532          server_id.host_port_pair());
533  scoped_ptr<Job> job(new Job(this, host_resolver_, host_port_pair, is_https,
534                              was_alternate_protocol_recently_broken,
535                              privacy_mode, method, quic_server_info, net_log));
536  int rv = job->Run(base::Bind(&QuicStreamFactory::OnJobComplete,
537                               base::Unretained(this), job.get()));
538
539  if (rv == ERR_IO_PENDING) {
540    active_requests_[request] = job.get();
541    job_requests_map_[job.get()].insert(request);
542    active_jobs_[server_id] = job.release();
543  }
544  if (rv == OK) {
545    DCHECK(HasActiveSession(server_id));
546    request->set_stream(CreateIfSessionExists(server_id, net_log));
547  }
548  return rv;
549}
550
551bool QuicStreamFactory::OnResolution(
552    const QuicServerId& server_id,
553    const AddressList& address_list) {
554  DCHECK(!HasActiveSession(server_id));
555  for (size_t i = 0; i < address_list.size(); ++i) {
556    const IPEndPoint& address = address_list[i];
557    const IpAliasKey ip_alias_key(address, server_id.is_https());
558    if (!ContainsKey(ip_aliases_, ip_alias_key))
559      continue;
560
561    const SessionSet& sessions = ip_aliases_[ip_alias_key];
562    for (SessionSet::const_iterator i = sessions.begin();
563         i != sessions.end(); ++i) {
564      QuicClientSession* session = *i;
565      if (!session->CanPool(server_id.host()))
566        continue;
567      active_sessions_[server_id] = session;
568      session_aliases_[session].insert(server_id);
569      return true;
570    }
571  }
572  return false;
573}
574
575void QuicStreamFactory::OnJobComplete(Job* job, int rv) {
576  if (rv == OK) {
577    require_confirmation_ = false;
578
579    // Create all the streams, but do not notify them yet.
580    for (RequestSet::iterator it = job_requests_map_[job].begin();
581         it != job_requests_map_[job].end() ; ++it) {
582      DCHECK(HasActiveSession(job->server_id()));
583      (*it)->set_stream(CreateIfSessionExists(job->server_id(),
584                                              (*it)->net_log()));
585    }
586  }
587  while (!job_requests_map_[job].empty()) {
588    RequestSet::iterator it = job_requests_map_[job].begin();
589    QuicStreamRequest* request = *it;
590    job_requests_map_[job].erase(it);
591    active_requests_.erase(request);
592    // Even though we're invoking callbacks here, we don't need to worry
593    // about |this| being deleted, because the factory is owned by the
594    // profile which can not be deleted via callbacks.
595    request->OnRequestComplete(rv);
596  }
597  active_jobs_.erase(job->server_id());
598  job_requests_map_.erase(job);
599  delete job;
600  return;
601}
602
603// Returns a newly created QuicHttpStream owned by the caller, if a
604// matching session already exists.  Returns NULL otherwise.
605scoped_ptr<QuicHttpStream> QuicStreamFactory::CreateIfSessionExists(
606    const QuicServerId& server_id,
607    const BoundNetLog& net_log) {
608  if (!HasActiveSession(server_id)) {
609    DVLOG(1) << "No active session";
610    return scoped_ptr<QuicHttpStream>();
611  }
612
613  QuicClientSession* session = active_sessions_[server_id];
614  DCHECK(session);
615  return scoped_ptr<QuicHttpStream>(
616      new QuicHttpStream(session->GetWeakPtr()));
617}
618
619void QuicStreamFactory::OnIdleSession(QuicClientSession* session) {
620}
621
622void QuicStreamFactory::OnSessionGoingAway(QuicClientSession* session) {
623  const AliasSet& aliases = session_aliases_[session];
624  for (AliasSet::const_iterator it = aliases.begin(); it != aliases.end();
625       ++it) {
626    DCHECK(active_sessions_.count(*it));
627    DCHECK_EQ(session, active_sessions_[*it]);
628    // Track sessions which have recently gone away so that we can disable
629    // port suggestions.
630    if (session->goaway_received()) {
631      gone_away_aliases_.insert(*it);
632    }
633
634    active_sessions_.erase(*it);
635    ProcessGoingAwaySession(session, *it, true);
636  }
637  ProcessGoingAwaySession(session, all_sessions_[session], false);
638  if (!aliases.empty()) {
639    const IpAliasKey ip_alias_key(session->connection()->peer_address(),
640                                  aliases.begin()->is_https());
641    ip_aliases_[ip_alias_key].erase(session);
642    if (ip_aliases_[ip_alias_key].empty()) {
643      ip_aliases_.erase(ip_alias_key);
644    }
645  }
646  session_aliases_.erase(session);
647}
648
649void QuicStreamFactory::OnSessionClosed(QuicClientSession* session) {
650  DCHECK_EQ(0u, session->GetNumOpenStreams());
651  OnSessionGoingAway(session);
652  delete session;
653  all_sessions_.erase(session);
654}
655
656void QuicStreamFactory::OnSessionConnectTimeout(
657    QuicClientSession* session) {
658  const AliasSet& aliases = session_aliases_[session];
659  for (AliasSet::const_iterator it = aliases.begin(); it != aliases.end();
660       ++it) {
661    DCHECK(active_sessions_.count(*it));
662    DCHECK_EQ(session, active_sessions_[*it]);
663    active_sessions_.erase(*it);
664  }
665
666  if (aliases.empty()) {
667    return;
668  }
669
670  const IpAliasKey ip_alias_key(session->connection()->peer_address(),
671                                aliases.begin()->is_https());
672  ip_aliases_[ip_alias_key].erase(session);
673  if (ip_aliases_[ip_alias_key].empty()) {
674    ip_aliases_.erase(ip_alias_key);
675  }
676  QuicServerId server_id = *aliases.begin();
677  session_aliases_.erase(session);
678  Job* job = new Job(this, host_resolver_, session, server_id);
679  active_jobs_[server_id] = job;
680  int rv = job->Run(base::Bind(&QuicStreamFactory::OnJobComplete,
681                               base::Unretained(this), job));
682  DCHECK_EQ(ERR_IO_PENDING, rv);
683}
684
685void QuicStreamFactory::CancelRequest(QuicStreamRequest* request) {
686  DCHECK(ContainsKey(active_requests_, request));
687  Job* job = active_requests_[request];
688  job_requests_map_[job].erase(request);
689  active_requests_.erase(request);
690}
691
692void QuicStreamFactory::CloseAllSessions(int error) {
693  while (!active_sessions_.empty()) {
694    size_t initial_size = active_sessions_.size();
695    active_sessions_.begin()->second->CloseSessionOnError(error);
696    DCHECK_NE(initial_size, active_sessions_.size());
697  }
698  while (!all_sessions_.empty()) {
699    size_t initial_size = all_sessions_.size();
700    all_sessions_.begin()->first->CloseSessionOnError(error);
701    DCHECK_NE(initial_size, all_sessions_.size());
702  }
703  DCHECK(all_sessions_.empty());
704}
705
706base::Value* QuicStreamFactory::QuicStreamFactoryInfoToValue() const {
707  base::ListValue* list = new base::ListValue();
708
709  for (SessionMap::const_iterator it = active_sessions_.begin();
710       it != active_sessions_.end(); ++it) {
711    const QuicServerId& server_id = it->first;
712    QuicClientSession* session = it->second;
713    const AliasSet& aliases = session_aliases_.find(session)->second;
714    // Only add a session to the list once.
715    if (server_id == *aliases.begin()) {
716      std::set<HostPortPair> hosts;
717      for (AliasSet::const_iterator alias_it = aliases.begin();
718           alias_it != aliases.end(); ++alias_it) {
719        hosts.insert(alias_it->host_port_pair());
720      }
721      list->Append(session->GetInfoAsValue(hosts));
722    }
723  }
724  return list;
725}
726
727void QuicStreamFactory::ClearCachedStatesInCryptoConfig() {
728  crypto_config_.ClearCachedStates();
729}
730
731void QuicStreamFactory::OnIPAddressChanged() {
732  CloseAllSessions(ERR_NETWORK_CHANGED);
733  require_confirmation_ = true;
734}
735
736void QuicStreamFactory::OnCertAdded(const X509Certificate* cert) {
737  CloseAllSessions(ERR_CERT_DATABASE_CHANGED);
738}
739
740void QuicStreamFactory::OnCACertChanged(const X509Certificate* cert) {
741  // We should flush the sessions if we removed trust from a
742  // cert, because a previously trusted server may have become
743  // untrusted.
744  //
745  // We should not flush the sessions if we added trust to a cert.
746  //
747  // Since the OnCACertChanged method doesn't tell us what
748  // kind of change it is, we have to flush the socket
749  // pools to be safe.
750  CloseAllSessions(ERR_CERT_DATABASE_CHANGED);
751}
752
753bool QuicStreamFactory::HasActiveSession(
754    const QuicServerId& server_id) const {
755  return ContainsKey(active_sessions_, server_id);
756}
757
758int QuicStreamFactory::CreateSession(
759    const QuicServerId& server_id,
760    scoped_ptr<QuicServerInfo> server_info,
761    const AddressList& address_list,
762    const BoundNetLog& net_log,
763    QuicClientSession** session) {
764  bool enable_port_selection = enable_port_selection_;
765  if (enable_port_selection &&
766      ContainsKey(gone_away_aliases_, server_id)) {
767    // Disable port selection when the server is going away.
768    // There is no point in trying to return to the same server, if
769    // that server is no longer handling requests.
770    enable_port_selection = false;
771    gone_away_aliases_.erase(server_id);
772  }
773
774  QuicConnectionId connection_id = random_generator_->RandUint64();
775  IPEndPoint addr = *address_list.begin();
776  scoped_refptr<PortSuggester> port_suggester =
777      new PortSuggester(server_id.host_port_pair(), port_seed_);
778  DatagramSocket::BindType bind_type = enable_port_selection ?
779      DatagramSocket::RANDOM_BIND :  // Use our callback.
780      DatagramSocket::DEFAULT_BIND;  // Use OS to randomize.
781  scoped_ptr<DatagramClientSocket> socket(
782      client_socket_factory_->CreateDatagramClientSocket(
783          bind_type,
784          base::Bind(&PortSuggester::SuggestPort, port_suggester),
785          net_log.net_log(), net_log.source()));
786  int rv = socket->Connect(addr);
787  if (rv != OK) {
788    HistogramCreateSessionFailure(CREATION_ERROR_CONNECTING_SOCKET);
789    return rv;
790  }
791  UMA_HISTOGRAM_COUNTS("Net.QuicEphemeralPortsSuggested",
792                       port_suggester->call_count());
793  if (enable_port_selection) {
794    DCHECK_LE(1u, port_suggester->call_count());
795  } else {
796    DCHECK_EQ(0u, port_suggester->call_count());
797  }
798
799  // We should adaptively set this buffer size, but for now, we'll use a size
800  // that is more than large enough for a full receive window, and yet
801  // does not consume "too much" memory.  If we see bursty packet loss, we may
802  // revisit this setting and test for its impact.
803  const int32 kSocketBufferSize(TcpReceiver::kReceiveWindowTCP);
804  rv = socket->SetReceiveBufferSize(kSocketBufferSize);
805  if (rv != OK) {
806    HistogramCreateSessionFailure(CREATION_ERROR_SETTING_RECEIVE_BUFFER);
807    return rv;
808  }
809  // Set a buffer large enough to contain the initial CWND's worth of packet
810  // to work around the problem with CHLO packets being sent out with the
811  // wrong encryption level, when the send buffer is full.
812  rv = socket->SetSendBufferSize(kMaxPacketSize * 20);
813  if (rv != OK) {
814    HistogramCreateSessionFailure(CREATION_ERROR_SETTING_SEND_BUFFER);
815    return rv;
816  }
817
818  scoped_ptr<QuicDefaultPacketWriter> writer(
819      new QuicDefaultPacketWriter(socket.get()));
820
821  if (!helper_.get()) {
822    helper_.reset(new QuicConnectionHelper(
823        base::MessageLoop::current()->message_loop_proxy().get(),
824        clock_.get(), random_generator_));
825  }
826
827  QuicConnection* connection =
828      new QuicConnection(connection_id, addr, helper_.get(), writer.get(),
829                         false, supported_versions_);
830  writer->SetConnection(connection);
831  connection->set_max_packet_length(max_packet_length_);
832
833  InitializeCachedStateInCryptoConfig(server_id, server_info);
834
835  QuicConfig config = config_;
836  config.SetInitialCongestionWindowToSend(
837      server_id.is_https() ? kServerSecureInitialCongestionWindow
838                           : kServerInecureInitialCongestionWindow);
839  config.SetInitialFlowControlWindowToSend(kInitialReceiveWindowSize);
840  config.SetInitialStreamFlowControlWindowToSend(kInitialReceiveWindowSize);
841  config.SetInitialSessionFlowControlWindowToSend(kInitialReceiveWindowSize);
842  if (http_server_properties_) {
843    const HttpServerProperties::NetworkStats* stats =
844        http_server_properties_->GetServerNetworkStats(
845            server_id.host_port_pair());
846    if (stats != NULL) {
847      config.SetInitialRoundTripTimeUsToSend(stats->srtt.InMicroseconds());
848    }
849  }
850
851  *session = new QuicClientSession(
852      connection, socket.Pass(), writer.Pass(), this,
853      quic_crypto_client_stream_factory_, server_info.Pass(), server_id,
854      config, &crypto_config_,
855      base::MessageLoop::current()->message_loop_proxy().get(),
856      net_log.net_log());
857  all_sessions_[*session] = server_id;  // owning pointer
858  return OK;
859}
860
861bool QuicStreamFactory::HasActiveJob(const QuicServerId& key) const {
862  return ContainsKey(active_jobs_, key);
863}
864
865void QuicStreamFactory::ActivateSession(
866    const QuicServerId& server_id,
867    QuicClientSession* session) {
868  DCHECK(!HasActiveSession(server_id));
869  UMA_HISTOGRAM_COUNTS("Net.QuicActiveSessions", active_sessions_.size());
870  active_sessions_[server_id] = session;
871  session_aliases_[session].insert(server_id);
872  const IpAliasKey ip_alias_key(session->connection()->peer_address(),
873                                server_id.is_https());
874  DCHECK(!ContainsKey(ip_aliases_[ip_alias_key], session));
875  ip_aliases_[ip_alias_key].insert(session);
876}
877
878void QuicStreamFactory::InitializeCachedStateInCryptoConfig(
879    const QuicServerId& server_id,
880    const scoped_ptr<QuicServerInfo>& server_info) {
881  if (!server_info)
882    return;
883
884  QuicCryptoClientConfig::CachedState* cached =
885      crypto_config_.LookupOrCreate(server_id);
886  if (!cached->IsEmpty())
887    return;
888
889  if (!cached->Initialize(server_info->state().server_config,
890                          server_info->state().source_address_token,
891                          server_info->state().certs,
892                          server_info->state().server_config_sig,
893                          clock_->WallNow()))
894    return;
895
896  if (!server_id.is_https()) {
897    // Don't check the certificates for insecure QUIC.
898    cached->SetProofValid();
899  }
900}
901
902void QuicStreamFactory::ProcessGoingAwaySession(
903    QuicClientSession* session,
904    const QuicServerId& server_id,
905    bool session_was_active) {
906  if (!http_server_properties_)
907    return;
908
909  const QuicConnectionStats& stats = session->connection()->GetStats();
910  if (session->IsCryptoHandshakeConfirmed()) {
911    HttpServerProperties::NetworkStats network_stats;
912    network_stats.srtt = base::TimeDelta::FromMicroseconds(stats.srtt_us);
913    network_stats.bandwidth_estimate = stats.estimated_bandwidth;
914    http_server_properties_->SetServerNetworkStats(server_id.host_port_pair(),
915                                                   network_stats);
916    return;
917  }
918
919  UMA_HISTOGRAM_COUNTS("Net.QuicHandshakeNotConfirmedNumPacketsReceived",
920                       stats.packets_received);
921
922  if (!session_was_active)
923    return;
924
925  const HostPortPair& server = server_id.host_port_pair();
926  // Don't try to change the alternate-protocol state, if the
927  // alternate-protocol state is unknown.
928  if (!http_server_properties_->HasAlternateProtocol(server))
929    return;
930
931  // TODO(rch):  In the special case where the session has received no
932  // packets from the peer, we should consider blacklisting this
933  // differently so that we still race TCP but we don't consider the
934  // session connected until the handshake has been confirmed.
935  HistogramBrokenAlternateProtocolLocation(
936      BROKEN_ALTERNATE_PROTOCOL_LOCATION_QUIC_STREAM_FACTORY);
937  PortAlternateProtocolPair alternate =
938      http_server_properties_->GetAlternateProtocol(server);
939  DCHECK_EQ(QUIC, alternate.protocol);
940
941  // Since the session was active, there's no longer an
942  // HttpStreamFactoryImpl::Job running which can mark it broken, unless the
943  // TCP job also fails. So to avoid not using QUIC when we otherwise could,
944  // we mark it as broken, and then immediately re-enable it. This leaves
945  // QUIC as "recently broken" which means that 0-RTT will be disabled but
946  // we'll still race.
947  http_server_properties_->SetBrokenAlternateProtocol(server);
948  http_server_properties_->ClearAlternateProtocol(server);
949  http_server_properties_->SetAlternateProtocol(
950      server, alternate.port, alternate.protocol);
951  DCHECK_EQ(QUIC,
952            http_server_properties_->GetAlternateProtocol(server).protocol);
953  DCHECK(http_server_properties_->WasAlternateProtocolRecentlyBroken(
954      server));
955}
956
957}  // namespace net
958