net/spdy/spdy_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/spdy/spdy_session.h"
   6
   7 #include <algorithm>
   8 #include <map>
   9
  10 #include "base/basictypes.h"
  11 #include "base/bind.h"
  12 #include "base/compiler_specific.h"
  13 #include "base/logging.h"
  14 #include "base/message_loop/message_loop.h"
  15 #include "base/metrics/field_trial.h"
  16 #include "base/metrics/histogram.h"
  17 #include "base/metrics/sparse_histogram.h"
  18 #include "base/metrics/stats_counters.h"
  19 #include "base/stl_util.h"
  20 #include "base/strings/string_number_conversions.h"
  21 #include "base/strings/string_util.h"
  22 #include "base/strings/stringprintf.h"
  23 #include "base/strings/utf_string_conversions.h"
  24 #include "base/time/time.h"
  25 #include "base/values.h"
  26 #include "crypto/ec_private_key.h"
  27 #include "crypto/ec_signature_creator.h"
  28 #include "net/base/connection_type_histograms.h"
  29 #include "net/base/net_log.h"
  30 #include "net/base/net_util.h"
  31 #include "net/cert/asn1_util.h"
  32 #include "net/cert/cert_verify_result.h"
  33 #include "net/http/http_log_util.h"
  34 #include "net/http/http_network_session.h"
  35 #include "net/http/http_server_properties.h"
  36 #include "net/http/http_util.h"
  37 #include "net/http/transport_security_state.h"
  38 #include "net/spdy/spdy_buffer_producer.h"
  39 #include "net/spdy/spdy_frame_builder.h"
  40 #include "net/spdy/spdy_http_utils.h"
  41 #include "net/spdy/spdy_protocol.h"
  42 #include "net/spdy/spdy_session_pool.h"
  43 #include "net/spdy/spdy_stream.h"
  44 #include "net/ssl/channel_id_service.h"
  45 #include "net/ssl/ssl_cipher_suite_names.h"
  46 #include "net/ssl/ssl_connection_status_flags.h"
  47
  48 namespace net {
  49
  50 namespace {
  51
  52 const int kReadBufferSize = 8 * 1024;
  53 const int kDefaultConnectionAtRiskOfLossSeconds = 10;
  54 const int kHungIntervalSeconds = 10;
  55
  56 // Minimum seconds that unclaimed pushed streams will be kept in memory.
  57 const int kMinPushedStreamLifetimeSeconds = 300;
  58
  59 scoped_ptr<base::ListValue> SpdyHeaderBlockToListValue(
  60     const SpdyHeaderBlock& headers,
  61     net::NetLog::LogLevel log_level) {
  62   scoped_ptr<base::ListValue> headers_list(new base::ListValue());
  63   for (SpdyHeaderBlock::const_iterator it = headers.begin();
  64        it != headers.end(); ++it) {
  65     headers_list->AppendString(
  66         it->first + ": " +
  67         ElideHeaderValueForNetLog(log_level, it->first, it->second));
  68   }
  69   return headers_list.Pass();
  70 }
  71
  72 base::Value* NetLogSpdySynStreamSentCallback(const SpdyHeaderBlock* headers,
  73                                              bool fin,
  74                                              bool unidirectional,
  75                                              SpdyPriority spdy_priority,
  76                                              SpdyStreamId stream_id,
  77                                              NetLog::LogLevel log_level) {
  78   base::DictionaryValue* dict = new base::DictionaryValue();
  79   dict->Set("headers",
  80             SpdyHeaderBlockToListValue(*headers, log_level).release());
  81   dict->SetBoolean("fin", fin);
  82   dict->SetBoolean("unidirectional", unidirectional);
  83   dict->SetInteger("spdy_priority", static_cast<int>(spdy_priority));
  84   dict->SetInteger("stream_id", stream_id);
  85   return dict;
  86 }
  87
  88 base::Value* NetLogSpdySynStreamReceivedCallback(
  89     const SpdyHeaderBlock* headers,
  90     bool fin,
  91     bool unidirectional,
  92     SpdyPriority spdy_priority,
  93     SpdyStreamId stream_id,
  94     SpdyStreamId associated_stream,
  95     NetLog::LogLevel log_level) {
  96   base::DictionaryValue* dict = new base::DictionaryValue();
  97   dict->Set("headers",
  98             SpdyHeaderBlockToListValue(*headers, log_level).release());
  99   dict->SetBoolean("fin", fin);
 100   dict->SetBoolean("unidirectional", unidirectional);
 101   dict->SetInteger("spdy_priority", static_cast<int>(spdy_priority));
 102   dict->SetInteger("stream_id", stream_id);
 103   dict->SetInteger("associated_stream", associated_stream);
 104   return dict;
 105 }
 106
 107 base::Value* NetLogSpdySynReplyOrHeadersReceivedCallback(
 108     const SpdyHeaderBlock* headers,
 109     bool fin,
 110     SpdyStreamId stream_id,
 111     NetLog::LogLevel log_level) {
 112   base::DictionaryValue* dict = new base::DictionaryValue();
 113   dict->Set("headers",
 114             SpdyHeaderBlockToListValue(*headers, log_level).release());
 115   dict->SetBoolean("fin", fin);
 116   dict->SetInteger("stream_id", stream_id);
 117   return dict;
 118 }
 119
 120 base::Value* NetLogSpdySessionCloseCallback(int net_error,
 121                                             const std::string* description,
 122                                             NetLog::LogLevel /* log_level */) {
 123   base::DictionaryValue* dict = new base::DictionaryValue();
 124   dict->SetInteger("net_error", net_error);
 125   dict->SetString("description", *description);
 126   return dict;
 127 }
 128
 129 base::Value* NetLogSpdySessionCallback(const HostPortProxyPair* host_pair,
 130                                        NetLog::LogLevel /* log_level */) {
 131   base::DictionaryValue* dict = new base::DictionaryValue();
 132   dict->SetString("host", host_pair->first.ToString());
 133   dict->SetString("proxy", host_pair->second.ToPacString());
 134   return dict;
 135 }
 136
 137 base::Value* NetLogSpdySettingsCallback(const HostPortPair& host_port_pair,
 138                                         bool clear_persisted,
 139                                         NetLog::LogLevel /* log_level */) {
 140   base::DictionaryValue* dict = new base::DictionaryValue();
 141   dict->SetString("host", host_port_pair.ToString());
 142   dict->SetBoolean("clear_persisted", clear_persisted);
 143   return dict;
 144 }
 145
 146 base::Value* NetLogSpdySettingCallback(SpdySettingsIds id,
 147                                        SpdySettingsFlags flags,
 148                                        uint32 value,
 149                                        NetLog::LogLevel /* log_level */) {
 150   base::DictionaryValue* dict = new base::DictionaryValue();
 151   dict->SetInteger("id", id);
 152   dict->SetInteger("flags", flags);
 153   dict->SetInteger("value", value);
 154   return dict;
 155 }
 156
 157 base::Value* NetLogSpdySendSettingsCallback(const SettingsMap* settings,
 158                                             NetLog::LogLevel /* log_level */) {
 159   base::DictionaryValue* dict = new base::DictionaryValue();
 160   base::ListValue* settings_list = new base::ListValue();
 161   for (SettingsMap::const_iterator it = settings->begin();
 162        it != settings->end(); ++it) {
 163     const SpdySettingsIds id = it->first;
 164     const SpdySettingsFlags flags = it->second.first;
 165     const uint32 value = it->second.second;
 166     settings_list->Append(new base::StringValue(
 167         base::StringPrintf("[id:%u flags:%u value:%u]", id, flags, value)));
 168   }
 169   dict->Set("settings", settings_list);
 170   return dict;
 171 }
 172
 173 base::Value* NetLogSpdyWindowUpdateFrameCallback(
 174     SpdyStreamId stream_id,
 175     uint32 delta,
 176     NetLog::LogLevel /* log_level */) {
 177   base::DictionaryValue* dict = new base::DictionaryValue();
 178   dict->SetInteger("stream_id", static_cast<int>(stream_id));
 179   dict->SetInteger("delta", delta);
 180   return dict;
 181 }
 182
 183 base::Value* NetLogSpdySessionWindowUpdateCallback(
 184     int32 delta,
 185     int32 window_size,
 186     NetLog::LogLevel /* log_level */) {
 187   base::DictionaryValue* dict = new base::DictionaryValue();
 188   dict->SetInteger("delta", delta);
 189   dict->SetInteger("window_size", window_size);
 190   return dict;
 191 }
 192
 193 base::Value* NetLogSpdyDataCallback(SpdyStreamId stream_id,
 194                                     int size,
 195                                     bool fin,
 196                                     NetLog::LogLevel /* log_level */) {
 197   base::DictionaryValue* dict = new base::DictionaryValue();
 198   dict->SetInteger("stream_id", static_cast<int>(stream_id));
 199   dict->SetInteger("size", size);
 200   dict->SetBoolean("fin", fin);
 201   return dict;
 202 }
 203
 204 base::Value* NetLogSpdyRstCallback(SpdyStreamId stream_id,
 205                                    int status,
 206                                    const std::string* description,
 207                                    NetLog::LogLevel /* log_level */) {
 208   base::DictionaryValue* dict = new base::DictionaryValue();
 209   dict->SetInteger("stream_id", static_cast<int>(stream_id));
 210   dict->SetInteger("status", status);
 211   dict->SetString("description", *description);
 212   return dict;
 213 }
 214
 215 base::Value* NetLogSpdyPingCallback(SpdyPingId unique_id,
 216                                     bool is_ack,
 217                                     const char* type,
 218                                     NetLog::LogLevel /* log_level */) {
 219   base::DictionaryValue* dict = new base::DictionaryValue();
 220   dict->SetInteger("unique_id", unique_id);
 221   dict->SetString("type", type);
 222   dict->SetBoolean("is_ack", is_ack);
 223   return dict;
 224 }
 225
 226 base::Value* NetLogSpdyGoAwayCallback(SpdyStreamId last_stream_id,
 227                                       int active_streams,
 228                                       int unclaimed_streams,
 229                                       SpdyGoAwayStatus status,
 230                                       NetLog::LogLevel /* log_level */) {
 231   base::DictionaryValue* dict = new base::DictionaryValue();
 232   dict->SetInteger("last_accepted_stream_id",
 233                    static_cast<int>(last_stream_id));
 234   dict->SetInteger("active_streams", active_streams);
 235   dict->SetInteger("unclaimed_streams", unclaimed_streams);
 236   dict->SetInteger("status", static_cast<int>(status));
 237   return dict;
 238 }
 239
 240 base::Value* NetLogSpdyPushPromiseReceivedCallback(
 241     const SpdyHeaderBlock* headers,
 242     SpdyStreamId stream_id,
 243     SpdyStreamId promised_stream_id,
 244     NetLog::LogLevel log_level) {
 245   base::DictionaryValue* dict = new base::DictionaryValue();
 246   dict->Set("headers",
 247             SpdyHeaderBlockToListValue(*headers, log_level).release());
 248   dict->SetInteger("id", stream_id);
 249   dict->SetInteger("promised_stream_id", promised_stream_id);
 250   return dict;
 251 }
 252
 253 // Helper function to return the total size of an array of objects
 254 // with .size() member functions.
 255 template <typename T, size_t N> size_t GetTotalSize(const T (&arr)[N]) {
 256   size_t total_size = 0;
 257   for (size_t i = 0; i < N; ++i) {
 258     total_size += arr[i].size();
 259   }
 260   return total_size;
 261 }
 262
 263 // Helper class for std:find_if on STL container containing
 264 // SpdyStreamRequest weak pointers.
 265 class RequestEquals {
 266  public:
 267   RequestEquals(const base::WeakPtr<SpdyStreamRequest>& request)
 268       : request_(request) {}
 269
 270   bool operator()(const base::WeakPtr<SpdyStreamRequest>& request) const {
 271     return request_.get() == request.get();
 272   }
 273
 274  private:
 275   const base::WeakPtr<SpdyStreamRequest> request_;
 276 };
 277
 278 // The maximum number of concurrent streams we will ever create.  Even if
 279 // the server permits more, we will never exceed this limit.
 280 const size_t kMaxConcurrentStreamLimit = 256;
 281
 282 }  // namespace
 283
 284 SpdyProtocolErrorDetails MapFramerErrorToProtocolError(
 285     SpdyFramer::SpdyError err) {
 286   switch(err) {
 287     case SpdyFramer::SPDY_NO_ERROR:
 288       return SPDY_ERROR_NO_ERROR;
 289     case SpdyFramer::SPDY_INVALID_CONTROL_FRAME:
 290       return SPDY_ERROR_INVALID_CONTROL_FRAME;
 291     case SpdyFramer::SPDY_CONTROL_PAYLOAD_TOO_LARGE:
 292       return SPDY_ERROR_CONTROL_PAYLOAD_TOO_LARGE;
 293     case SpdyFramer::SPDY_ZLIB_INIT_FAILURE:
 294       return SPDY_ERROR_ZLIB_INIT_FAILURE;
 295     case SpdyFramer::SPDY_UNSUPPORTED_VERSION:
 296       return SPDY_ERROR_UNSUPPORTED_VERSION;
 297     case SpdyFramer::SPDY_DECOMPRESS_FAILURE:
 298       return SPDY_ERROR_DECOMPRESS_FAILURE;
 299     case SpdyFramer::SPDY_COMPRESS_FAILURE:
 300       return SPDY_ERROR_COMPRESS_FAILURE;
 301     case SpdyFramer::SPDY_GOAWAY_FRAME_CORRUPT:
 302       return SPDY_ERROR_GOAWAY_FRAME_CORRUPT;
 303     case SpdyFramer::SPDY_RST_STREAM_FRAME_CORRUPT:
 304       return SPDY_ERROR_RST_STREAM_FRAME_CORRUPT;
 305     case SpdyFramer::SPDY_INVALID_DATA_FRAME_FLAGS:
 306       return SPDY_ERROR_INVALID_DATA_FRAME_FLAGS;
 307     case SpdyFramer::SPDY_INVALID_CONTROL_FRAME_FLAGS:
 308       return SPDY_ERROR_INVALID_CONTROL_FRAME_FLAGS;
 309     case SpdyFramer::SPDY_UNEXPECTED_FRAME:
 310       return SPDY_ERROR_UNEXPECTED_FRAME;
 311     default:
 312       NOTREACHED();
 313       return static_cast<SpdyProtocolErrorDetails>(-1);
 314   }
 315 }
 316
 317 Error MapFramerErrorToNetError(SpdyFramer::SpdyError err) {
 318   switch (err) {
 319     case SpdyFramer::SPDY_NO_ERROR:
 320       return OK;
 321     case SpdyFramer::SPDY_INVALID_CONTROL_FRAME:
 322       return ERR_SPDY_PROTOCOL_ERROR;
 323     case SpdyFramer::SPDY_CONTROL_PAYLOAD_TOO_LARGE:
 324       return ERR_SPDY_FRAME_SIZE_ERROR;
 325     case SpdyFramer::SPDY_ZLIB_INIT_FAILURE:
 326       return ERR_SPDY_COMPRESSION_ERROR;
 327     case SpdyFramer::SPDY_UNSUPPORTED_VERSION:
 328       return ERR_SPDY_PROTOCOL_ERROR;
 329     case SpdyFramer::SPDY_DECOMPRESS_FAILURE:
 330       return ERR_SPDY_COMPRESSION_ERROR;
 331     case SpdyFramer::SPDY_COMPRESS_FAILURE:
 332       return ERR_SPDY_COMPRESSION_ERROR;
 333     case SpdyFramer::SPDY_GOAWAY_FRAME_CORRUPT:
 334       return ERR_SPDY_PROTOCOL_ERROR;
 335     case SpdyFramer::SPDY_RST_STREAM_FRAME_CORRUPT:
 336       return ERR_SPDY_PROTOCOL_ERROR;
 337     case SpdyFramer::SPDY_INVALID_DATA_FRAME_FLAGS:
 338       return ERR_SPDY_PROTOCOL_ERROR;
 339     case SpdyFramer::SPDY_INVALID_CONTROL_FRAME_FLAGS:
 340       return ERR_SPDY_PROTOCOL_ERROR;
 341     case SpdyFramer::SPDY_UNEXPECTED_FRAME:
 342       return ERR_SPDY_PROTOCOL_ERROR;
 343     default:
 344       NOTREACHED();
 345       return ERR_SPDY_PROTOCOL_ERROR;
 346   }
 347 }
 348
 349 SpdyProtocolErrorDetails MapRstStreamStatusToProtocolError(
 350     SpdyRstStreamStatus status) {
 351   switch(status) {
 352     case RST_STREAM_PROTOCOL_ERROR:
 353       return STATUS_CODE_PROTOCOL_ERROR;
 354     case RST_STREAM_INVALID_STREAM:
 355       return STATUS_CODE_INVALID_STREAM;
 356     case RST_STREAM_REFUSED_STREAM:
 357       return STATUS_CODE_REFUSED_STREAM;
 358     case RST_STREAM_UNSUPPORTED_VERSION:
 359       return STATUS_CODE_UNSUPPORTED_VERSION;
 360     case RST_STREAM_CANCEL:
 361       return STATUS_CODE_CANCEL;
 362     case RST_STREAM_INTERNAL_ERROR:
 363       return STATUS_CODE_INTERNAL_ERROR;
 364     case RST_STREAM_FLOW_CONTROL_ERROR:
 365       return STATUS_CODE_FLOW_CONTROL_ERROR;
 366     case RST_STREAM_STREAM_IN_USE:
 367       return STATUS_CODE_STREAM_IN_USE;
 368     case RST_STREAM_STREAM_ALREADY_CLOSED:
 369       return STATUS_CODE_STREAM_ALREADY_CLOSED;
 370     case RST_STREAM_INVALID_CREDENTIALS:
 371       return STATUS_CODE_INVALID_CREDENTIALS;
 372     case RST_STREAM_FRAME_SIZE_ERROR:
 373       return STATUS_CODE_FRAME_SIZE_ERROR;
 374     case RST_STREAM_SETTINGS_TIMEOUT:
 375       return STATUS_CODE_SETTINGS_TIMEOUT;
 376     case RST_STREAM_CONNECT_ERROR:
 377       return STATUS_CODE_CONNECT_ERROR;
 378     case RST_STREAM_ENHANCE_YOUR_CALM:
 379       return STATUS_CODE_ENHANCE_YOUR_CALM;
 380     default:
 381       NOTREACHED();
 382       return static_cast<SpdyProtocolErrorDetails>(-1);
 383   }
 384 }
 385
 386 SpdyGoAwayStatus MapNetErrorToGoAwayStatus(Error err) {
 387   switch (err) {
 388     case OK:
 389       return GOAWAY_NO_ERROR;
 390     case ERR_SPDY_PROTOCOL_ERROR:
 391       return GOAWAY_PROTOCOL_ERROR;
 392     case ERR_SPDY_FLOW_CONTROL_ERROR:
 393       return GOAWAY_FLOW_CONTROL_ERROR;
 394     case ERR_SPDY_FRAME_SIZE_ERROR:
 395       return GOAWAY_FRAME_SIZE_ERROR;
 396     case ERR_SPDY_COMPRESSION_ERROR:
 397       return GOAWAY_COMPRESSION_ERROR;
 398     case ERR_SPDY_INADEQUATE_TRANSPORT_SECURITY:
 399       return GOAWAY_INADEQUATE_SECURITY;
 400     default:
 401       return GOAWAY_PROTOCOL_ERROR;
 402   }
 403 }
 404
 405 void SplitPushedHeadersToRequestAndResponse(const SpdyHeaderBlock& headers,
 406                                             SpdyMajorVersion protocol_version,
 407                                             SpdyHeaderBlock* request_headers,
 408                                             SpdyHeaderBlock* response_headers) {
 409   DCHECK(response_headers);
 410   DCHECK(request_headers);
 411   for (SpdyHeaderBlock::const_iterator it = headers.begin();
 412        it != headers.end();
 413        ++it) {
 414     SpdyHeaderBlock* to_insert = response_headers;
 415     if (protocol_version == SPDY2) {
 416       if (it->first == "url")
 417         to_insert = request_headers;
 418     } else {
 419       const char* host = protocol_version >= SPDY4 ? ":authority" : ":host";
 420       static const char* scheme = ":scheme";
 421       static const char* path = ":path";
 422       if (it->first == host || it->first == scheme || it->first == path)
 423         to_insert = request_headers;
 424     }
 425     to_insert->insert(*it);
 426   }
 427 }
 428
 429 SpdyStreamRequest::SpdyStreamRequest() : weak_ptr_factory_(this) {
 430   Reset();
 431 }
 432
 433 SpdyStreamRequest::~SpdyStreamRequest() {
 434   CancelRequest();
 435 }
 436
 437 int SpdyStreamRequest::StartRequest(
 438     SpdyStreamType type,
 439     const base::WeakPtr<SpdySession>& session,
 440     const GURL& url,
 441     RequestPriority priority,
 442     const BoundNetLog& net_log,
 443     const CompletionCallback& callback) {
 444   DCHECK(session);
 445   DCHECK(!session_);
 446   DCHECK(!stream_);
 447   DCHECK(callback_.is_null());
 448
 449   type_ = type;
 450   session_ = session;
 451   url_ = url;
 452   priority_ = priority;
 453   net_log_ = net_log;
 454   callback_ = callback;
 455
 456   base::WeakPtr<SpdyStream> stream;
 457   int rv = session->TryCreateStream(weak_ptr_factory_.GetWeakPtr(), &stream);
 458   if (rv == OK) {
 459     Reset();
 460     stream_ = stream;
 461   }
 462   return rv;
 463 }
 464
 465 void SpdyStreamRequest::CancelRequest() {
 466   if (session_)
 467     session_->CancelStreamRequest(weak_ptr_factory_.GetWeakPtr());
 468   Reset();
 469   // Do this to cancel any pending CompleteStreamRequest() tasks.
 470   weak_ptr_factory_.InvalidateWeakPtrs();
 471 }
 472
 473 base::WeakPtr<SpdyStream> SpdyStreamRequest::ReleaseStream() {
 474   DCHECK(!session_);
 475   base::WeakPtr<SpdyStream> stream = stream_;
 476   DCHECK(stream);
 477   Reset();
 478   return stream;
 479 }
 480
 481 void SpdyStreamRequest::OnRequestCompleteSuccess(
 482     const base::WeakPtr<SpdyStream>& stream) {
 483   DCHECK(session_);
 484   DCHECK(!stream_);
 485   DCHECK(!callback_.is_null());
 486   CompletionCallback callback = callback_;
 487   Reset();
 488   DCHECK(stream);
 489   stream_ = stream;
 490   callback.Run(OK);
 491 }
 492
 493 void SpdyStreamRequest::OnRequestCompleteFailure(int rv) {
 494   DCHECK(session_);
 495   DCHECK(!stream_);
 496   DCHECK(!callback_.is_null());
 497   CompletionCallback callback = callback_;
 498   Reset();
 499   DCHECK_NE(rv, OK);
 500   callback.Run(rv);
 501 }
 502
 503 void SpdyStreamRequest::Reset() {
 504   type_ = SPDY_BIDIRECTIONAL_STREAM;
 505   session_.reset();
 506   stream_.reset();
 507   url_ = GURL();
 508   priority_ = MINIMUM_PRIORITY;
 509   net_log_ = BoundNetLog();
 510   callback_.Reset();
 511 }
 512
 513 SpdySession::ActiveStreamInfo::ActiveStreamInfo()
 514     : stream(NULL),
 515       waiting_for_syn_reply(false) {}
 516
 517 SpdySession::ActiveStreamInfo::ActiveStreamInfo(SpdyStream* stream)
 518     : stream(stream),
 519       waiting_for_syn_reply(stream->type() != SPDY_PUSH_STREAM) {
 520 }
 521
 522 SpdySession::ActiveStreamInfo::~ActiveStreamInfo() {}
 523
 524 SpdySession::PushedStreamInfo::PushedStreamInfo() : stream_id(0) {}
 525
 526 SpdySession::PushedStreamInfo::PushedStreamInfo(
 527     SpdyStreamId stream_id,
 528     base::TimeTicks creation_time)
 529     : stream_id(stream_id),
 530       creation_time(creation_time) {}
 531
 532 SpdySession::PushedStreamInfo::~PushedStreamInfo() {}
 533
 534 // static
 535 bool SpdySession::CanPool(TransportSecurityState* transport_security_state,
 536                           const SSLInfo& ssl_info,
 537                           const std::string& old_hostname,
 538                           const std::string& new_hostname) {
 539   // Pooling is prohibited if the server cert is not valid for the new domain,
 540   // and for connections on which client certs were sent. It is also prohibited
 541   // when channel ID was sent if the hosts are from different eTLDs+1.
 542   if (IsCertStatusError(ssl_info.cert_status))
 543     return false;
 544
 545   if (ssl_info.client_cert_sent)
 546     return false;
 547
 548   if (ssl_info.channel_id_sent &&
 549       ChannelIDService::GetDomainForHost(new_hostname) !=
 550       ChannelIDService::GetDomainForHost(old_hostname)) {
 551     return false;
 552   }
 553
 554   bool unused = false;
 555   if (!ssl_info.cert->VerifyNameMatch(new_hostname, &unused))
 556     return false;
 557
 558   std::string pinning_failure_log;
 559   if (!transport_security_state->CheckPublicKeyPins(
 560           new_hostname,
 561           true, /* sni_available */
 562           ssl_info.is_issued_by_known_root,
 563           ssl_info.public_key_hashes,
 564           &pinning_failure_log)) {
 565     return false;
 566   }
 567
 568   return true;
 569 }
 570
 571 SpdySession::SpdySession(
 572     const SpdySessionKey& spdy_session_key,
 573     const base::WeakPtr<HttpServerProperties>& http_server_properties,
 574     TransportSecurityState* transport_security_state,
 575     bool verify_domain_authentication,
 576     bool enable_sending_initial_data,
 577     bool enable_compression,
 578     bool enable_ping_based_connection_checking,
 579     NextProto default_protocol,
 580     size_t stream_initial_recv_window_size,
 581     size_t initial_max_concurrent_streams,
 582     size_t max_concurrent_streams_limit,
 583     TimeFunc time_func,
 584     const HostPortPair& trusted_spdy_proxy,
 585     NetLog* net_log)
 586     : in_io_loop_(false),
 587       spdy_session_key_(spdy_session_key),
 588       pool_(NULL),
 589       http_server_properties_(http_server_properties),
 590       transport_security_state_(transport_security_state),
 591       read_buffer_(new IOBuffer(kReadBufferSize)),
 592       stream_hi_water_mark_(kFirstStreamId),
 593       last_accepted_push_stream_id_(0),
 594       num_pushed_streams_(0u),
 595       num_active_pushed_streams_(0u),
 596       in_flight_write_frame_type_(DATA),
 597       in_flight_write_frame_size_(0),
 598       is_secure_(false),
 599       certificate_error_code_(OK),
 600       availability_state_(STATE_AVAILABLE),
 601       read_state_(READ_STATE_DO_READ),
 602       write_state_(WRITE_STATE_IDLE),
 603       error_on_close_(OK),
 604       max_concurrent_streams_(initial_max_concurrent_streams == 0
 605                                   ? kInitialMaxConcurrentStreams
 606                                   : initial_max_concurrent_streams),
 607       max_concurrent_streams_limit_(max_concurrent_streams_limit == 0
 608                                         ? kMaxConcurrentStreamLimit
 609                                         : max_concurrent_streams_limit),
 610       max_concurrent_pushed_streams_(kMaxConcurrentPushedStreams),
 611       streams_initiated_count_(0),
 612       streams_pushed_count_(0),
 613       streams_pushed_and_claimed_count_(0),
 614       streams_abandoned_count_(0),
 615       total_bytes_received_(0),
 616       sent_settings_(false),
 617       received_settings_(false),
 618       stalled_streams_(0),
 619       pings_in_flight_(0),
 620       next_ping_id_(1),
 621       last_activity_time_(time_func()),
 622       last_compressed_frame_len_(0),
 623       check_ping_status_pending_(false),
 624       send_connection_header_prefix_(false),
 625       flow_control_state_(FLOW_CONTROL_NONE),
 626       stream_initial_send_window_size_(kSpdyStreamInitialWindowSize),
 627       stream_initial_recv_window_size_(stream_initial_recv_window_size == 0
 628                                            ? kDefaultInitialRecvWindowSize
 629                                            : stream_initial_recv_window_size),
 630       session_send_window_size_(0),
 631       session_recv_window_size_(0),
 632       session_unacked_recv_window_bytes_(0),
 633       net_log_(BoundNetLog::Make(net_log, NetLog::SOURCE_SPDY_SESSION)),
 634       verify_domain_authentication_(verify_domain_authentication),
 635       enable_sending_initial_data_(enable_sending_initial_data),
 636       enable_compression_(enable_compression),
 637       enable_ping_based_connection_checking_(
 638           enable_ping_based_connection_checking),
 639       protocol_(default_protocol),
 640       connection_at_risk_of_loss_time_(
 641           base::TimeDelta::FromSeconds(kDefaultConnectionAtRiskOfLossSeconds)),
 642       hung_interval_(base::TimeDelta::FromSeconds(kHungIntervalSeconds)),
 643       trusted_spdy_proxy_(trusted_spdy_proxy),
 644       time_func_(time_func),
 645       weak_factory_(this) {
 646   DCHECK_GE(protocol_, kProtoSPDYMinimumVersion);
 647   DCHECK_LE(protocol_, kProtoSPDYMaximumVersion);
 648   DCHECK(HttpStreamFactory::spdy_enabled());
 649   net_log_.BeginEvent(
 650       NetLog::TYPE_SPDY_SESSION,
 651       base::Bind(&NetLogSpdySessionCallback, &host_port_proxy_pair()));
 652   next_unclaimed_push_stream_sweep_time_ = time_func_() +
 653       base::TimeDelta::FromSeconds(kMinPushedStreamLifetimeSeconds);
 654   // TODO(mbelshe): consider randomization of the stream_hi_water_mark.
 655 }
 656
 657 SpdySession::~SpdySession() {
 658   CHECK(!in_io_loop_);
 659   DcheckDraining();
 660
 661   // TODO(akalin): Check connection->is_initialized() instead. This
 662   // requires re-working CreateFakeSpdySession(), though.
 663   DCHECK(connection_->socket());
 664   // With SPDY we can't recycle sockets.
 665   connection_->socket()->Disconnect();
 666
 667   RecordHistograms();
 668
 669   net_log_.EndEvent(NetLog::TYPE_SPDY_SESSION);
 670 }
 671
 672 void SpdySession::InitializeWithSocket(
 673     scoped_ptr<ClientSocketHandle> connection,
 674     SpdySessionPool* pool,
 675     bool is_secure,
 676     int certificate_error_code) {
 677   CHECK(!in_io_loop_);
 678   DCHECK_EQ(availability_state_, STATE_AVAILABLE);
 679   DCHECK_EQ(read_state_, READ_STATE_DO_READ);
 680   DCHECK_EQ(write_state_, WRITE_STATE_IDLE);
 681   DCHECK(!connection_);
 682
 683   DCHECK(certificate_error_code == OK ||
 684          certificate_error_code < ERR_IO_PENDING);
 685   // TODO(akalin): Check connection->is_initialized() instead. This
 686   // requires re-working CreateFakeSpdySession(), though.
 687   DCHECK(connection->socket());
 688
 689   base::StatsCounter spdy_sessions("spdy.sessions");
 690   spdy_sessions.Increment();
 691
 692   connection_ = connection.Pass();
 693   is_secure_ = is_secure;
 694   certificate_error_code_ = certificate_error_code;
 695
 696   NextProto protocol_negotiated =
 697       connection_->socket()->GetNegotiatedProtocol();
 698   if (protocol_negotiated != kProtoUnknown) {
 699     protocol_ = protocol_negotiated;
 700   }
 701   DCHECK_GE(protocol_, kProtoSPDYMinimumVersion);
 702   DCHECK_LE(protocol_, kProtoSPDYMaximumVersion);
 703
 704   if (protocol_ == kProtoSPDY4)
 705     send_connection_header_prefix_ = true;
 706
 707   if (protocol_ >= kProtoSPDY31) {
 708     flow_control_state_ = FLOW_CONTROL_STREAM_AND_SESSION;
 709     session_send_window_size_ = kSpdySessionInitialWindowSize;
 710     session_recv_window_size_ = kSpdySessionInitialWindowSize;
 711   } else if (protocol_ >= kProtoSPDY3) {
 712     flow_control_state_ = FLOW_CONTROL_STREAM;
 713   } else {
 714     flow_control_state_ = FLOW_CONTROL_NONE;
 715   }
 716
 717   buffered_spdy_framer_.reset(
 718       new BufferedSpdyFramer(NextProtoToSpdyMajorVersion(protocol_),
 719                              enable_compression_));
 720   buffered_spdy_framer_->set_visitor(this);
 721   buffered_spdy_framer_->set_debug_visitor(this);
 722   UMA_HISTOGRAM_ENUMERATION("Net.SpdyVersion", protocol_, kProtoMaximumVersion);
 723 #if defined(SPDY_PROXY_AUTH_ORIGIN)
 724   UMA_HISTOGRAM_BOOLEAN("Net.SpdySessions_DataReductionProxy",
 725                         host_port_pair().Equals(HostPortPair::FromURL(
 726                             GURL(SPDY_PROXY_AUTH_ORIGIN))));
 727 #endif
 728
 729   net_log_.AddEvent(
 730       NetLog::TYPE_SPDY_SESSION_INITIALIZED,
 731       connection_->socket()->NetLog().source().ToEventParametersCallback());
 732
 733   DCHECK_EQ(availability_state_, STATE_AVAILABLE);
 734   connection_->AddHigherLayeredPool(this);
 735   if (enable_sending_initial_data_)
 736     SendInitialData();
 737   pool_ = pool;
 738
 739   // Bootstrap the read loop.
 740   base::MessageLoop::current()->PostTask(
 741       FROM_HERE,
 742       base::Bind(&SpdySession::PumpReadLoop,
 743                  weak_factory_.GetWeakPtr(), READ_STATE_DO_READ, OK));
 744 }
 745
 746 bool SpdySession::VerifyDomainAuthentication(const std::string& domain) {
 747   if (!verify_domain_authentication_)
 748     return true;
 749
 750   if (availability_state_ == STATE_DRAINING)
 751     return false;
 752
 753   SSLInfo ssl_info;
 754   bool was_npn_negotiated;
 755   NextProto protocol_negotiated = kProtoUnknown;
 756   if (!GetSSLInfo(&ssl_info, &was_npn_negotiated, &protocol_negotiated))
 757     return true;   // This is not a secure session, so all domains are okay.
 758
 759   return CanPool(transport_security_state_, ssl_info,
 760                  host_port_pair().host(), domain);
 761 }
 762
 763 int SpdySession::GetPushStream(
 764     const GURL& url,
 765     base::WeakPtr<SpdyStream>* stream,
 766     const BoundNetLog& stream_net_log) {
 767   CHECK(!in_io_loop_);
 768
 769   stream->reset();
 770
 771   if (availability_state_ == STATE_DRAINING)
 772     return ERR_CONNECTION_CLOSED;
 773
 774   Error err = TryAccessStream(url);
 775   if (err != OK)
 776     return err;
 777
 778   *stream = GetActivePushStream(url);
 779   if (*stream) {
 780     DCHECK_LT(streams_pushed_and_claimed_count_, streams_pushed_count_);
 781     streams_pushed_and_claimed_count_++;
 782   }
 783   return OK;
 784 }
 785
 786 // {,Try}CreateStream() and TryAccessStream() can be called with
 787 // |in_io_loop_| set if a stream is being created in response to
 788 // another being closed due to received data.
 789
 790 Error SpdySession::TryAccessStream(const GURL& url) {
 791   if (is_secure_ && certificate_error_code_ != OK &&
 792       (url.SchemeIs("https") || url.SchemeIs("wss"))) {
 793     RecordProtocolErrorHistogram(
 794         PROTOCOL_ERROR_REQUEST_FOR_SECURE_CONTENT_OVER_INSECURE_SESSION);
 795     DoDrainSession(
 796         static_cast<Error>(certificate_error_code_),
 797         "Tried to get SPDY stream for secure content over an unauthenticated "
 798         "session.");
 799     return ERR_SPDY_PROTOCOL_ERROR;
 800   }
 801   return OK;
 802 }
 803
 804 int SpdySession::TryCreateStream(
 805     const base::WeakPtr<SpdyStreamRequest>& request,
 806     base::WeakPtr<SpdyStream>* stream) {
 807   DCHECK(request);
 808
 809   if (availability_state_ == STATE_GOING_AWAY)
 810     return ERR_FAILED;
 811
 812   if (availability_state_ == STATE_DRAINING)
 813     return ERR_CONNECTION_CLOSED;
 814
 815   Error err = TryAccessStream(request->url());
 816   if (err != OK)
 817     return err;
 818
 819   if (!max_concurrent_streams_ ||
 820       (active_streams_.size() + created_streams_.size() - num_pushed_streams_ <
 821        max_concurrent_streams_)) {
 822     return CreateStream(*request, stream);
 823   }
 824
 825   stalled_streams_++;
 826   net_log().AddEvent(NetLog::TYPE_SPDY_SESSION_STALLED_MAX_STREAMS);
 827   RequestPriority priority = request->priority();
 828   CHECK_GE(priority, MINIMUM_PRIORITY);
 829   CHECK_LE(priority, MAXIMUM_PRIORITY);
 830   pending_create_stream_queues_[priority].push_back(request);
 831   return ERR_IO_PENDING;
 832 }
 833
 834 int SpdySession::CreateStream(const SpdyStreamRequest& request,
 835                               base::WeakPtr<SpdyStream>* stream) {
 836   DCHECK_GE(request.priority(), MINIMUM_PRIORITY);
 837   DCHECK_LE(request.priority(), MAXIMUM_PRIORITY);
 838
 839   if (availability_state_ == STATE_GOING_AWAY)
 840     return ERR_FAILED;
 841
 842   if (availability_state_ == STATE_DRAINING)
 843     return ERR_CONNECTION_CLOSED;
 844
 845   Error err = TryAccessStream(request.url());
 846   if (err != OK) {
 847     // This should have been caught in TryCreateStream().
 848     NOTREACHED();
 849     return err;
 850   }
 851
 852   DCHECK(connection_->socket());
 853   DCHECK(connection_->socket()->IsConnected());
 854   if (connection_->socket()) {
 855     UMA_HISTOGRAM_BOOLEAN("Net.SpdySession.CreateStreamWithSocketConnected",
 856                           connection_->socket()->IsConnected());
 857     if (!connection_->socket()->IsConnected()) {
 858       DoDrainSession(
 859           ERR_CONNECTION_CLOSED,
 860           "Tried to create SPDY stream for a closed socket connection.");
 861       return ERR_CONNECTION_CLOSED;
 862     }
 863   }
 864
 865   scoped_ptr<SpdyStream> new_stream(
 866       new SpdyStream(request.type(), GetWeakPtr(), request.url(),
 867                      request.priority(),
 868                      stream_initial_send_window_size_,
 869                      stream_initial_recv_window_size_,
 870                      request.net_log()));
 871   *stream = new_stream->GetWeakPtr();
 872   InsertCreatedStream(new_stream.Pass());
 873
 874   UMA_HISTOGRAM_CUSTOM_COUNTS(
 875       "Net.SpdyPriorityCount",
 876       static_cast<int>(request.priority()), 0, 10, 11);
 877
 878   return OK;
 879 }
 880
 881 void SpdySession::CancelStreamRequest(
 882     const base::WeakPtr<SpdyStreamRequest>& request) {
 883   DCHECK(request);
 884   RequestPriority priority = request->priority();
 885   CHECK_GE(priority, MINIMUM_PRIORITY);
 886   CHECK_LE(priority, MAXIMUM_PRIORITY);
 887
 888 #if DCHECK_IS_ON
 889   // |request| should not be in a queue not matching its priority.
 890   for (int i = MINIMUM_PRIORITY; i <= MAXIMUM_PRIORITY; ++i) {
 891     if (priority == i)
 892       continue;
 893     PendingStreamRequestQueue* queue = &pending_create_stream_queues_[i];
 894     DCHECK(std::find_if(queue->begin(),
 895                         queue->end(),
 896                         RequestEquals(request)) == queue->end());
 897   }
 898 #endif
 899
 900   PendingStreamRequestQueue* queue =
 901       &pending_create_stream_queues_[priority];
 902   // Remove |request| from |queue| while preserving the order of the
 903   // other elements.
 904   PendingStreamRequestQueue::iterator it =
 905       std::find_if(queue->begin(), queue->end(), RequestEquals(request));
 906   // The request may already be removed if there's a
 907   // CompleteStreamRequest() in flight.
 908   if (it != queue->end()) {
 909     it = queue->erase(it);
 910     // |request| should be in the queue at most once, and if it is
 911     // present, should not be pending completion.
 912     DCHECK(std::find_if(it, queue->end(), RequestEquals(request)) ==
 913            queue->end());
 914   }
 915 }
 916
 917 base::WeakPtr<SpdyStreamRequest> SpdySession::GetNextPendingStreamRequest() {
 918   for (int j = MAXIMUM_PRIORITY; j >= MINIMUM_PRIORITY; --j) {
 919     if (pending_create_stream_queues_[j].empty())
 920       continue;
 921
 922     base::WeakPtr<SpdyStreamRequest> pending_request =
 923         pending_create_stream_queues_[j].front();
 924     DCHECK(pending_request);
 925     pending_create_stream_queues_[j].pop_front();
 926     return pending_request;
 927   }
 928   return base::WeakPtr<SpdyStreamRequest>();
 929 }
 930
 931 void SpdySession::ProcessPendingStreamRequests() {
 932   // Like |max_concurrent_streams_|, 0 means infinite for
 933   // |max_requests_to_process|.
 934   size_t max_requests_to_process = 0;
 935   if (max_concurrent_streams_ != 0) {
 936     max_requests_to_process =
 937         max_concurrent_streams_ -
 938         (active_streams_.size() + created_streams_.size());
 939   }
 940   for (size_t i = 0;
 941        max_requests_to_process == 0 || i < max_requests_to_process; ++i) {
 942     base::WeakPtr<SpdyStreamRequest> pending_request =
 943         GetNextPendingStreamRequest();
 944     if (!pending_request)
 945       break;
 946
 947     // Note that this post can race with other stream creations, and it's
 948     // possible that the un-stalled stream will be stalled again if it loses.
 949     // TODO(jgraettinger): Provide stronger ordering guarantees.
 950     base::MessageLoop::current()->PostTask(
 951         FROM_HERE,
 952         base::Bind(&SpdySession::CompleteStreamRequest,
 953                    weak_factory_.GetWeakPtr(),
 954                    pending_request));
 955   }
 956 }
 957
 958 void SpdySession::AddPooledAlias(const SpdySessionKey& alias_key) {
 959   pooled_aliases_.insert(alias_key);
 960 }
 961
 962 SpdyMajorVersion SpdySession::GetProtocolVersion() const {
 963   DCHECK(buffered_spdy_framer_.get());
 964   return buffered_spdy_framer_->protocol_version();
 965 }
 966
 967 bool SpdySession::HasAcceptableTransportSecurity() const {
 968   // If we're not even using TLS, we have no standards to meet.
 969   if (!is_secure_) {
 970     return true;
 971   }
 972
 973   // We don't enforce transport security standards for older SPDY versions.
 974   if (GetProtocolVersion() < SPDY4) {
 975     return true;
 976   }
 977
 978   SSLInfo ssl_info;
 979   CHECK(connection_->socket()->GetSSLInfo(&ssl_info));
 980
 981   // HTTP/2 requires TLS 1.2+
 982   if (SSLConnectionStatusToVersion(ssl_info.connection_status) <
 983       SSL_CONNECTION_VERSION_TLS1_2) {
 984     return false;
 985   }
 986
 987   if (!IsSecureTLSCipherSuite(
 988           SSLConnectionStatusToCipherSuite(ssl_info.connection_status))) {
 989     return false;
 990   }
 991
 992   return true;
 993 }
 994
 995 base::WeakPtr<SpdySession> SpdySession::GetWeakPtr() {
 996   return weak_factory_.GetWeakPtr();
 997 }
 998
 999 bool SpdySession::CloseOneIdleConnection() {
1000   CHECK(!in_io_loop_);
1001   DCHECK(pool_);
1002   if (active_streams_.empty()) {
1003     DoDrainSession(ERR_CONNECTION_CLOSED, "Closing idle connection.");
1004   }
1005   // Return false as the socket wasn't immediately closed.
1006   return false;
1007 }
1008
1009 void SpdySession::EnqueueStreamWrite(
1010     const base::WeakPtr<SpdyStream>& stream,
1011     SpdyFrameType frame_type,
1012     scoped_ptr<SpdyBufferProducer> producer) {
1013   DCHECK(frame_type == HEADERS ||
1014          frame_type == DATA ||
1015          frame_type == CREDENTIAL ||
1016          frame_type == SYN_STREAM);
1017   EnqueueWrite(stream->priority(), frame_type, producer.Pass(), stream);
1018 }
1019
1020 scoped_ptr<SpdyFrame> SpdySession::CreateSynStream(
1021     SpdyStreamId stream_id,
1022     RequestPriority priority,
1023     SpdyControlFlags flags,
1024     const SpdyHeaderBlock& block) {
1025   ActiveStreamMap::const_iterator it = active_streams_.find(stream_id);
1026   CHECK(it != active_streams_.end());
1027   CHECK_EQ(it->second.stream->stream_id(), stream_id);
1028
1029   SendPrefacePingIfNoneInFlight();
1030
1031   DCHECK(buffered_spdy_framer_.get());
1032   SpdyPriority spdy_priority =
1033       ConvertRequestPriorityToSpdyPriority(priority, GetProtocolVersion());
1034
1035   scoped_ptr<SpdyFrame> syn_frame;
1036   // TODO(hkhalil): Avoid copy of |block|.
1037   if (GetProtocolVersion() <= SPDY3) {
1038     SpdySynStreamIR syn_stream(stream_id);
1039     syn_stream.set_associated_to_stream_id(0);
1040     syn_stream.set_priority(spdy_priority);
1041     syn_stream.set_fin((flags & CONTROL_FLAG_FIN) != 0);
1042     syn_stream.set_unidirectional((flags & CONTROL_FLAG_UNIDIRECTIONAL) != 0);
1043     syn_stream.set_name_value_block(block);
1044     syn_frame.reset(buffered_spdy_framer_->SerializeFrame(syn_stream));
1045   } else {
1046     SpdyHeadersIR headers(stream_id);
1047     headers.set_priority(spdy_priority);
1048     headers.set_has_priority(true);
1049     headers.set_fin((flags & CONTROL_FLAG_FIN) != 0);
1050     headers.set_name_value_block(block);
1051     syn_frame.reset(buffered_spdy_framer_->SerializeFrame(headers));
1052   }
1053
1054   base::StatsCounter spdy_requests("spdy.requests");
1055   spdy_requests.Increment();
1056   streams_initiated_count_++;
1057
1058   if (net_log().IsLogging()) {
1059     net_log().AddEvent(NetLog::TYPE_SPDY_SESSION_SYN_STREAM,
1060                        base::Bind(&NetLogSpdySynStreamSentCallback,
1061                                   &block,
1062                                   (flags & CONTROL_FLAG_FIN) != 0,
1063                                   (flags & CONTROL_FLAG_UNIDIRECTIONAL) != 0,
1064                                   spdy_priority,
1065                                   stream_id));
1066   }
1067
1068   return syn_frame.Pass();
1069 }
1070
1071 scoped_ptr<SpdyBuffer> SpdySession::CreateDataBuffer(SpdyStreamId stream_id,
1072                                                      IOBuffer* data,
1073                                                      int len,
1074                                                      SpdyDataFlags flags) {
1075   if (availability_state_ == STATE_DRAINING) {
1076     return scoped_ptr<SpdyBuffer>();
1077   }
1078
1079   ActiveStreamMap::const_iterator it = active_streams_.find(stream_id);
1080   CHECK(it != active_streams_.end());
1081   SpdyStream* stream = it->second.stream;
1082   CHECK_EQ(stream->stream_id(), stream_id);
1083
1084   if (len < 0) {
1085     NOTREACHED();
1086     return scoped_ptr<SpdyBuffer>();
1087   }
1088
1089   int effective_len = std::min(len, kMaxSpdyFrameChunkSize);
1090
1091   bool send_stalled_by_stream =
1092       (flow_control_state_ >= FLOW_CONTROL_STREAM) &&
1093       (stream->send_window_size() <= 0);
1094   bool send_stalled_by_session = IsSendStalled();
1095
1096   // NOTE: There's an enum of the same name in histograms.xml.
1097   enum SpdyFrameFlowControlState {
1098     SEND_NOT_STALLED,
1099     SEND_STALLED_BY_STREAM,
1100     SEND_STALLED_BY_SESSION,
1101     SEND_STALLED_BY_STREAM_AND_SESSION,
1102   };
1103
1104   SpdyFrameFlowControlState frame_flow_control_state = SEND_NOT_STALLED;
1105   if (send_stalled_by_stream) {
1106     if (send_stalled_by_session) {
1107       frame_flow_control_state = SEND_STALLED_BY_STREAM_AND_SESSION;
1108     } else {
1109       frame_flow_control_state = SEND_STALLED_BY_STREAM;
1110     }
1111   } else if (send_stalled_by_session) {
1112     frame_flow_control_state = SEND_STALLED_BY_SESSION;
1113   }
1114
1115   if (flow_control_state_ == FLOW_CONTROL_STREAM) {
1116     UMA_HISTOGRAM_ENUMERATION(
1117         "Net.SpdyFrameStreamFlowControlState",
1118         frame_flow_control_state,
1119         SEND_STALLED_BY_STREAM + 1);
1120   } else if (flow_control_state_ == FLOW_CONTROL_STREAM_AND_SESSION) {
1121     UMA_HISTOGRAM_ENUMERATION(
1122         "Net.SpdyFrameStreamAndSessionFlowControlState",
1123         frame_flow_control_state,
1124         SEND_STALLED_BY_STREAM_AND_SESSION + 1);
1125   }
1126
1127   // Obey send window size of the stream if stream flow control is
1128   // enabled.
1129   if (flow_control_state_ >= FLOW_CONTROL_STREAM) {
1130     if (send_stalled_by_stream) {
1131       stream->set_send_stalled_by_flow_control(true);
1132       // Even though we're currently stalled only by the stream, we
1133       // might end up being stalled by the session also.
1134       QueueSendStalledStream(*stream);
1135       net_log().AddEvent(
1136           NetLog::TYPE_SPDY_SESSION_STREAM_STALLED_BY_STREAM_SEND_WINDOW,
1137           NetLog::IntegerCallback("stream_id", stream_id));
1138       return scoped_ptr<SpdyBuffer>();
1139     }
1140
1141     effective_len = std::min(effective_len, stream->send_window_size());
1142   }
1143
1144   // Obey send window size of the session if session flow control is
1145   // enabled.
1146   if (flow_control_state_ == FLOW_CONTROL_STREAM_AND_SESSION) {
1147     if (send_stalled_by_session) {
1148       stream->set_send_stalled_by_flow_control(true);
1149       QueueSendStalledStream(*stream);
1150       net_log().AddEvent(
1151           NetLog::TYPE_SPDY_SESSION_STREAM_STALLED_BY_SESSION_SEND_WINDOW,
1152           NetLog::IntegerCallback("stream_id", stream_id));
1153       return scoped_ptr<SpdyBuffer>();
1154     }
1155
1156     effective_len = std::min(effective_len, session_send_window_size_);
1157   }
1158
1159   DCHECK_GE(effective_len, 0);
1160
1161   // Clear FIN flag if only some of the data will be in the data
1162   // frame.
1163   if (effective_len < len)
1164     flags = static_cast<SpdyDataFlags>(flags & ~DATA_FLAG_FIN);
1165
1166   if (net_log().IsLogging()) {
1167     net_log().AddEvent(
1168         NetLog::TYPE_SPDY_SESSION_SEND_DATA,
1169         base::Bind(&NetLogSpdyDataCallback, stream_id, effective_len,
1170                    (flags & DATA_FLAG_FIN) != 0));
1171   }
1172
1173   // Send PrefacePing for DATA_FRAMEs with nonzero payload size.
1174   if (effective_len > 0)
1175     SendPrefacePingIfNoneInFlight();
1176
1177   // TODO(mbelshe): reduce memory copies here.
1178   DCHECK(buffered_spdy_framer_.get());
1179   scoped_ptr<SpdyFrame> frame(
1180       buffered_spdy_framer_->CreateDataFrame(
1181           stream_id, data->data(),
1182           static_cast<uint32>(effective_len), flags));
1183
1184   scoped_ptr<SpdyBuffer> data_buffer(new SpdyBuffer(frame.Pass()));
1185
1186   // Send window size is based on payload size, so nothing to do if this is
1187   // just a FIN with no payload.
1188   if (flow_control_state_ == FLOW_CONTROL_STREAM_AND_SESSION &&
1189       effective_len != 0) {
1190     DecreaseSendWindowSize(static_cast<int32>(effective_len));
1191     data_buffer->AddConsumeCallback(
1192         base::Bind(&SpdySession::OnWriteBufferConsumed,
1193                    weak_factory_.GetWeakPtr(),
1194                    static_cast<size_t>(effective_len)));
1195   }
1196
1197   return data_buffer.Pass();
1198 }
1199
1200 void SpdySession::CloseActiveStream(SpdyStreamId stream_id, int status) {
1201   DCHECK_NE(stream_id, 0u);
1202
1203   ActiveStreamMap::iterator it = active_streams_.find(stream_id);
1204   if (it == active_streams_.end()) {
1205     NOTREACHED();
1206     return;
1207   }
1208
1209   CloseActiveStreamIterator(it, status);
1210 }
1211
1212 void SpdySession::CloseCreatedStream(
1213     const base::WeakPtr<SpdyStream>& stream, int status) {
1214   DCHECK_EQ(stream->stream_id(), 0u);
1215
1216   CreatedStreamSet::iterator it = created_streams_.find(stream.get());
1217   if (it == created_streams_.end()) {
1218     NOTREACHED();
1219     return;
1220   }
1221
1222   CloseCreatedStreamIterator(it, status);
1223 }
1224
1225 void SpdySession::ResetStream(SpdyStreamId stream_id,
1226                               SpdyRstStreamStatus status,
1227                               const std::string& description) {
1228   DCHECK_NE(stream_id, 0u);
1229
1230   ActiveStreamMap::iterator it = active_streams_.find(stream_id);
1231   if (it == active_streams_.end()) {
1232     NOTREACHED();
1233     return;
1234   }
1235
1236   ResetStreamIterator(it, status, description);
1237 }
1238
1239 bool SpdySession::IsStreamActive(SpdyStreamId stream_id) const {
1240   return ContainsKey(active_streams_, stream_id);
1241 }
1242
1243 LoadState SpdySession::GetLoadState() const {
1244   // Just report that we're idle since the session could be doing
1245   // many things concurrently.
1246   return LOAD_STATE_IDLE;
1247 }
1248
1249 void SpdySession::CloseActiveStreamIterator(ActiveStreamMap::iterator it,
1250                                             int status) {
1251   // TODO(mbelshe): We should send a RST_STREAM control frame here
1252   //                so that the server can cancel a large send.
1253
1254   scoped_ptr<SpdyStream> owned_stream(it->second.stream);
1255   active_streams_.erase(it);
1256
1257   // TODO(akalin): When SpdyStream was ref-counted (and
1258   // |unclaimed_pushed_streams_| held scoped_refptr<SpdyStream>), this
1259   // was only done when status was not OK. This meant that pushed
1260   // streams can still be claimed after they're closed. This is
1261   // probably something that we still want to support, although server
1262   // push is hardly used. Write tests for this and fix this. (See
1263   // http://crbug.com/261712 .)
1264   if (owned_stream->type() == SPDY_PUSH_STREAM) {
1265     unclaimed_pushed_streams_.erase(owned_stream->url());
1266     num_pushed_streams_--;
1267     if (!owned_stream->IsReservedRemote())
1268       num_active_pushed_streams_--;
1269   }
1270
1271   DeleteStream(owned_stream.Pass(), status);
1272   MaybeFinishGoingAway();
1273
1274   // If there are no active streams and the socket pool is stalled, close the
1275   // session to free up a socket slot.
1276   if (active_streams_.empty() && connection_->IsPoolStalled()) {
1277     DoDrainSession(ERR_CONNECTION_CLOSED, "Closing idle connection.");
1278   }
1279 }
1280
1281 void SpdySession::CloseCreatedStreamIterator(CreatedStreamSet::iterator it,
1282                                              int status) {
1283   scoped_ptr<SpdyStream> owned_stream(*it);
1284   created_streams_.erase(it);
1285   DeleteStream(owned_stream.Pass(), status);
1286 }
1287
1288 void SpdySession::ResetStreamIterator(ActiveStreamMap::iterator it,
1289                                       SpdyRstStreamStatus status,
1290                                       const std::string& description) {
1291   // Send the RST_STREAM frame first as CloseActiveStreamIterator()
1292   // may close us.
1293   SpdyStreamId stream_id = it->first;
1294   RequestPriority priority = it->second.stream->priority();
1295   EnqueueResetStreamFrame(stream_id, priority, status, description);
1296
1297   // Removes any pending writes for the stream except for possibly an
1298   // in-flight one.
1299   CloseActiveStreamIterator(it, ERR_SPDY_PROTOCOL_ERROR);
1300 }
1301
1302 void SpdySession::EnqueueResetStreamFrame(SpdyStreamId stream_id,
1303                                           RequestPriority priority,
1304                                           SpdyRstStreamStatus status,
1305                                           const std::string& description) {
1306   DCHECK_NE(stream_id, 0u);
1307
1308   net_log().AddEvent(
1309       NetLog::TYPE_SPDY_SESSION_SEND_RST_STREAM,
1310       base::Bind(&NetLogSpdyRstCallback, stream_id, status, &description));
1311
1312   DCHECK(buffered_spdy_framer_.get());
1313   scoped_ptr<SpdyFrame> rst_frame(
1314       buffered_spdy_framer_->CreateRstStream(stream_id, status));
1315
1316   EnqueueSessionWrite(priority, RST_STREAM, rst_frame.Pass());
1317   RecordProtocolErrorHistogram(MapRstStreamStatusToProtocolError(status));
1318 }
1319
1320 void SpdySession::PumpReadLoop(ReadState expected_read_state, int result) {
1321   CHECK(!in_io_loop_);
1322   if (availability_state_ == STATE_DRAINING) {
1323     return;
1324   }
1325   ignore_result(DoReadLoop(expected_read_state, result));
1326 }
1327
1328 int SpdySession::DoReadLoop(ReadState expected_read_state, int result) {
1329   CHECK(!in_io_loop_);
1330   CHECK_EQ(read_state_, expected_read_state);
1331
1332   in_io_loop_ = true;
1333
1334   int bytes_read_without_yielding = 0;
1335
1336   // Loop until the session is draining, the read becomes blocked, or
1337   // the read limit is exceeded.
1338   while (true) {
1339     switch (read_state_) {
1340       case READ_STATE_DO_READ:
1341         CHECK_EQ(result, OK);
1342         result = DoRead();
1343         break;
1344       case READ_STATE_DO_READ_COMPLETE:
1345         if (result > 0)
1346           bytes_read_without_yielding += result;
1347         result = DoReadComplete(result);
1348         break;
1349       default:
1350         NOTREACHED() << "read_state_: " << read_state_;
1351         break;
1352     }
1353
1354     if (availability_state_ == STATE_DRAINING)
1355       break;
1356
1357     if (result == ERR_IO_PENDING)
1358       break;
1359
1360     if (bytes_read_without_yielding > kMaxReadBytesWithoutYielding) {
1361       read_state_ = READ_STATE_DO_READ;
1362       base::MessageLoop::current()->PostTask(
1363           FROM_HERE,
1364           base::Bind(&SpdySession::PumpReadLoop,
1365                      weak_factory_.GetWeakPtr(), READ_STATE_DO_READ, OK));
1366       result = ERR_IO_PENDING;
1367       break;
1368     }
1369   }
1370
1371   CHECK(in_io_loop_);
1372   in_io_loop_ = false;
1373
1374   return result;
1375 }
1376
1377 int SpdySession::DoRead() {
1378   CHECK(in_io_loop_);
1379
1380   CHECK(connection_);
1381   CHECK(connection_->socket());
1382   read_state_ = READ_STATE_DO_READ_COMPLETE;
1383   return connection_->socket()->Read(
1384       read_buffer_.get(),
1385       kReadBufferSize,
1386       base::Bind(&SpdySession::PumpReadLoop,
1387                  weak_factory_.GetWeakPtr(), READ_STATE_DO_READ_COMPLETE));
1388 }
1389
1390 int SpdySession::DoReadComplete(int result) {
1391   CHECK(in_io_loop_);
1392
1393   // Parse a frame.  For now this code requires that the frame fit into our
1394   // buffer (kReadBufferSize).
1395   // TODO(mbelshe): support arbitrarily large frames!
1396
1397   if (result == 0) {
1398     UMA_HISTOGRAM_CUSTOM_COUNTS("Net.SpdySession.BytesRead.EOF",
1399                                 total_bytes_received_, 1, 100000000, 50);
1400     DoDrainSession(ERR_CONNECTION_CLOSED, "Connection closed");
1401
1402     return ERR_CONNECTION_CLOSED;
1403   }
1404
1405   if (result < 0) {
1406     DoDrainSession(static_cast<Error>(result), "result is < 0.");
1407     return result;
1408   }
1409   CHECK_LE(result, kReadBufferSize);
1410   total_bytes_received_ += result;
1411
1412   last_activity_time_ = time_func_();
1413
1414   DCHECK(buffered_spdy_framer_.get());
1415   char* data = read_buffer_->data();
1416   while (result > 0) {
1417     uint32 bytes_processed = buffered_spdy_framer_->ProcessInput(data, result);
1418     result -= bytes_processed;
1419     data += bytes_processed;
1420
1421     if (availability_state_ == STATE_DRAINING) {
1422       return ERR_CONNECTION_CLOSED;
1423     }
1424
1425     DCHECK_EQ(buffered_spdy_framer_->error_code(), SpdyFramer::SPDY_NO_ERROR);
1426   }
1427
1428   read_state_ = READ_STATE_DO_READ;
1429   return OK;
1430 }
1431
1432 void SpdySession::PumpWriteLoop(WriteState expected_write_state, int result) {
1433   CHECK(!in_io_loop_);
1434   DCHECK_EQ(write_state_, expected_write_state);
1435
1436   DoWriteLoop(expected_write_state, result);
1437
1438   if (availability_state_ == STATE_DRAINING && !in_flight_write_ &&
1439       write_queue_.IsEmpty()) {
1440     pool_->RemoveUnavailableSession(GetWeakPtr());  // Destroys |this|.
1441     return;
1442   }
1443 }
1444
1445 int SpdySession::DoWriteLoop(WriteState expected_write_state, int result) {
1446   CHECK(!in_io_loop_);
1447   DCHECK_NE(write_state_, WRITE_STATE_IDLE);
1448   DCHECK_EQ(write_state_, expected_write_state);
1449
1450   in_io_loop_ = true;
1451
1452   // Loop until the session is closed or the write becomes blocked.
1453   while (true) {
1454     switch (write_state_) {
1455       case WRITE_STATE_DO_WRITE:
1456         DCHECK_EQ(result, OK);
1457         result = DoWrite();
1458         break;
1459       case WRITE_STATE_DO_WRITE_COMPLETE:
1460         result = DoWriteComplete(result);
1461         break;
1462       case WRITE_STATE_IDLE:
1463       default:
1464         NOTREACHED() << "write_state_: " << write_state_;
1465         break;
1466     }
1467
1468     if (write_state_ == WRITE_STATE_IDLE) {
1469       DCHECK_EQ(result, ERR_IO_PENDING);
1470       break;
1471     }
1472
1473     if (result == ERR_IO_PENDING)
1474       break;
1475   }
1476
1477   CHECK(in_io_loop_);
1478   in_io_loop_ = false;
1479
1480   return result;
1481 }
1482
1483 int SpdySession::DoWrite() {
1484   CHECK(in_io_loop_);
1485
1486   DCHECK(buffered_spdy_framer_);
1487   if (in_flight_write_) {
1488     DCHECK_GT(in_flight_write_->GetRemainingSize(), 0u);
1489   } else {
1490     // Grab the next frame to send.
1491     SpdyFrameType frame_type = DATA;
1492     scoped_ptr<SpdyBufferProducer> producer;
1493     base::WeakPtr<SpdyStream> stream;
1494     if (!write_queue_.Dequeue(&frame_type, &producer, &stream)) {
1495       write_state_ = WRITE_STATE_IDLE;
1496       return ERR_IO_PENDING;
1497     }
1498
1499     if (stream.get())
1500       CHECK(!stream->IsClosed());
1501
1502     // Activate the stream only when sending the SYN_STREAM frame to
1503     // guarantee monotonically-increasing stream IDs.
1504     if (frame_type == SYN_STREAM) {
1505       CHECK(stream.get());
1506       CHECK_EQ(stream->stream_id(), 0u);
1507       scoped_ptr<SpdyStream> owned_stream =
1508           ActivateCreatedStream(stream.get());
1509       InsertActivatedStream(owned_stream.Pass());
1510
1511       if (stream_hi_water_mark_ > kLastStreamId) {
1512         CHECK_EQ(stream->stream_id(), kLastStreamId);
1513         // We've exhausted the stream ID space, and no new streams may be
1514         // created after this one.
1515         MakeUnavailable();
1516         StartGoingAway(kLastStreamId, ERR_ABORTED);
1517       }
1518     }
1519
1520     in_flight_write_ = producer->ProduceBuffer();
1521     if (!in_flight_write_) {
1522       NOTREACHED();
1523       return ERR_UNEXPECTED;
1524     }
1525     in_flight_write_frame_type_ = frame_type;
1526     in_flight_write_frame_size_ = in_flight_write_->GetRemainingSize();
1527     DCHECK_GE(in_flight_write_frame_size_,
1528               buffered_spdy_framer_->GetFrameMinimumSize());
1529     in_flight_write_stream_ = stream;
1530   }
1531
1532   write_state_ = WRITE_STATE_DO_WRITE_COMPLETE;
1533
1534   // Explicitly store in a scoped_refptr<IOBuffer> to avoid problems
1535   // with Socket implementations that don't store their IOBuffer
1536   // argument in a scoped_refptr<IOBuffer> (see crbug.com/232345).
1537   scoped_refptr<IOBuffer> write_io_buffer =
1538       in_flight_write_->GetIOBufferForRemainingData();
1539   return connection_->socket()->Write(
1540       write_io_buffer.get(),
1541       in_flight_write_->GetRemainingSize(),
1542       base::Bind(&SpdySession::PumpWriteLoop,
1543                  weak_factory_.GetWeakPtr(), WRITE_STATE_DO_WRITE_COMPLETE));
1544 }
1545
1546 int SpdySession::DoWriteComplete(int result) {
1547   CHECK(in_io_loop_);
1548   DCHECK_NE(result, ERR_IO_PENDING);
1549   DCHECK_GT(in_flight_write_->GetRemainingSize(), 0u);
1550
1551   last_activity_time_ = time_func_();
1552
1553   if (result < 0) {
1554     DCHECK_NE(result, ERR_IO_PENDING);
1555     in_flight_write_.reset();
1556     in_flight_write_frame_type_ = DATA;
1557     in_flight_write_frame_size_ = 0;
1558     in_flight_write_stream_.reset();
1559     write_state_ = WRITE_STATE_DO_WRITE;
1560     DoDrainSession(static_cast<Error>(result), "Write error");
1561     return OK;
1562   }
1563
1564   // It should not be possible to have written more bytes than our
1565   // in_flight_write_.
1566   DCHECK_LE(static_cast<size_t>(result),
1567             in_flight_write_->GetRemainingSize());
1568
1569   if (result > 0) {
1570     in_flight_write_->Consume(static_cast<size_t>(result));
1571
1572     // We only notify the stream when we've fully written the pending frame.
1573     if (in_flight_write_->GetRemainingSize() == 0) {
1574       // It is possible that the stream was cancelled while we were
1575       // writing to the socket.
1576       if (in_flight_write_stream_.get()) {
1577         DCHECK_GT(in_flight_write_frame_size_, 0u);
1578         in_flight_write_stream_->OnFrameWriteComplete(
1579             in_flight_write_frame_type_,
1580             in_flight_write_frame_size_);
1581       }
1582
1583       // Cleanup the write which just completed.
1584       in_flight_write_.reset();
1585       in_flight_write_frame_type_ = DATA;
1586       in_flight_write_frame_size_ = 0;
1587       in_flight_write_stream_.reset();
1588     }
1589   }
1590
1591   write_state_ = WRITE_STATE_DO_WRITE;
1592   return OK;
1593 }
1594
1595 void SpdySession::DcheckGoingAway() const {
1596 #if DCHECK_IS_ON
1597   DCHECK_GE(availability_state_, STATE_GOING_AWAY);
1598   for (int i = MINIMUM_PRIORITY; i <= MAXIMUM_PRIORITY; ++i) {
1599     DCHECK(pending_create_stream_queues_[i].empty());
1600   }
1601   DCHECK(created_streams_.empty());
1602 #endif
1603 }
1604
1605 void SpdySession::DcheckDraining() const {
1606   DcheckGoingAway();
1607   DCHECK_EQ(availability_state_, STATE_DRAINING);
1608   DCHECK(active_streams_.empty());
1609   DCHECK(unclaimed_pushed_streams_.empty());
1610 }
1611
1612 void SpdySession::StartGoingAway(SpdyStreamId last_good_stream_id,
1613                                  Error status) {
1614   DCHECK_GE(availability_state_, STATE_GOING_AWAY);
1615
1616   // The loops below are carefully written to avoid reentrancy problems.
1617
1618   while (true) {
1619     size_t old_size = GetTotalSize(pending_create_stream_queues_);
1620     base::WeakPtr<SpdyStreamRequest> pending_request =
1621         GetNextPendingStreamRequest();
1622     if (!pending_request)
1623       break;
1624     // No new stream requests should be added while the session is
1625     // going away.
1626     DCHECK_GT(old_size, GetTotalSize(pending_create_stream_queues_));
1627     pending_request->OnRequestCompleteFailure(ERR_ABORTED);
1628   }
1629
1630   while (true) {
1631     size_t old_size = active_streams_.size();
1632     ActiveStreamMap::iterator it =
1633         active_streams_.lower_bound(last_good_stream_id + 1);
1634     if (it == active_streams_.end())
1635       break;
1636     LogAbandonedActiveStream(it, status);
1637     CloseActiveStreamIterator(it, status);
1638     // No new streams should be activated while the session is going
1639     // away.
1640     DCHECK_GT(old_size, active_streams_.size());
1641   }
1642
1643   while (!created_streams_.empty()) {
1644     size_t old_size = created_streams_.size();
1645     CreatedStreamSet::iterator it = created_streams_.begin();
1646     LogAbandonedStream(*it, status);
1647     CloseCreatedStreamIterator(it, status);
1648     // No new streams should be created while the session is going
1649     // away.
1650     DCHECK_GT(old_size, created_streams_.size());
1651   }
1652
1653   write_queue_.RemovePendingWritesForStreamsAfter(last_good_stream_id);
1654
1655   DcheckGoingAway();
1656 }
1657
1658 void SpdySession::MaybeFinishGoingAway() {
1659   if (active_streams_.empty() && availability_state_ == STATE_GOING_AWAY) {
1660     DoDrainSession(OK, "Finished going away");
1661   }
1662 }
1663
1664 void SpdySession::DoDrainSession(Error err, const std::string& description) {
1665   if (availability_state_ == STATE_DRAINING) {
1666     return;
1667   }
1668   MakeUnavailable();
1669
1670   // If |err| indicates an error occurred, inform the peer that we're closing
1671   // and why. Don't GOAWAY on a graceful or idle close, as that may
1672   // unnecessarily wake the radio. We could technically GOAWAY on network errors
1673   // (we'll probably fail to actually write it, but that's okay), however many
1674   // unit-tests would need to be updated.
1675   if (err != OK &&
1676       err != ERR_ABORTED &&  // Used by SpdySessionPool to close idle sessions.
1677       err != ERR_NETWORK_CHANGED &&  // Used to deprecate sessions on IP change.
1678       err != ERR_SOCKET_NOT_CONNECTED &&
1679       err != ERR_CONNECTION_CLOSED && err != ERR_CONNECTION_RESET) {
1680     // Enqueue a GOAWAY to inform the peer of why we're closing the connection.
1681     SpdyGoAwayIR goaway_ir(last_accepted_push_stream_id_,
1682                            MapNetErrorToGoAwayStatus(err),
1683                            description);
1684     EnqueueSessionWrite(HIGHEST,
1685                         GOAWAY,
1686                         scoped_ptr<SpdyFrame>(
1687                             buffered_spdy_framer_->SerializeFrame(goaway_ir)));
1688   }
1689
1690   availability_state_ = STATE_DRAINING;
1691   error_on_close_ = err;
1692
1693   net_log_.AddEvent(
1694       NetLog::TYPE_SPDY_SESSION_CLOSE,
1695       base::Bind(&NetLogSpdySessionCloseCallback, err, &description));
1696
1697   UMA_HISTOGRAM_SPARSE_SLOWLY("Net.SpdySession.ClosedOnError", -err);
1698   UMA_HISTOGRAM_CUSTOM_COUNTS("Net.SpdySession.BytesRead.OtherErrors",
1699                               total_bytes_received_, 1, 100000000, 50);
1700
1701   if (err == OK) {
1702     // We ought to be going away already, as this is a graceful close.
1703     DcheckGoingAway();
1704   } else {
1705     StartGoingAway(0, err);
1706   }
1707   DcheckDraining();
1708   MaybePostWriteLoop();
1709 }
1710
1711 void SpdySession::LogAbandonedStream(SpdyStream* stream, Error status) {
1712   DCHECK(stream);
1713   std::string description = base::StringPrintf(
1714       "ABANDONED (stream_id=%d): ", stream->stream_id()) +
1715       stream->url().spec();
1716   stream->LogStreamError(status, description);
1717   // We don't increment the streams abandoned counter here. If the
1718   // stream isn't active (i.e., it hasn't written anything to the wire
1719   // yet) then it's as if it never existed. If it is active, then
1720   // LogAbandonedActiveStream() will increment the counters.
1721 }
1722
1723 void SpdySession::LogAbandonedActiveStream(ActiveStreamMap::const_iterator it,
1724                                            Error status) {
1725   DCHECK_GT(it->first, 0u);
1726   LogAbandonedStream(it->second.stream, status);
1727   ++streams_abandoned_count_;
1728   base::StatsCounter abandoned_streams("spdy.abandoned_streams");
1729   abandoned_streams.Increment();
1730   if (it->second.stream->type() == SPDY_PUSH_STREAM &&
1731       unclaimed_pushed_streams_.find(it->second.stream->url()) !=
1732       unclaimed_pushed_streams_.end()) {
1733     base::StatsCounter abandoned_push_streams("spdy.abandoned_push_streams");
1734     abandoned_push_streams.Increment();
1735   }
1736 }
1737
1738 SpdyStreamId SpdySession::GetNewStreamId() {
1739   CHECK_LE(stream_hi_water_mark_, kLastStreamId);
1740   SpdyStreamId id = stream_hi_water_mark_;
1741   stream_hi_water_mark_ += 2;
1742   return id;
1743 }
1744
1745 void SpdySession::CloseSessionOnError(Error err,
1746                                       const std::string& description) {
1747   DCHECK_LT(err, ERR_IO_PENDING);
1748   DoDrainSession(err, description);
1749 }
1750
1751 void SpdySession::MakeUnavailable() {
1752   if (availability_state_ == STATE_AVAILABLE) {
1753     availability_state_ = STATE_GOING_AWAY;
1754     pool_->MakeSessionUnavailable(GetWeakPtr());
1755   }
1756 }
1757
1758 base::Value* SpdySession::GetInfoAsValue() const {
1759   base::DictionaryValue* dict = new base::DictionaryValue();
1760
1761   dict->SetInteger("source_id", net_log_.source().id);
1762
1763   dict->SetString("host_port_pair", host_port_pair().ToString());
1764   if (!pooled_aliases_.empty()) {
1765     base::ListValue* alias_list = new base::ListValue();
1766     for (std::set<SpdySessionKey>::const_iterator it =
1767              pooled_aliases_.begin();
1768          it != pooled_aliases_.end(); it++) {
1769       alias_list->Append(new base::StringValue(
1770           it->host_port_pair().ToString()));
1771     }
1772     dict->Set("aliases", alias_list);
1773   }
1774   dict->SetString("proxy", host_port_proxy_pair().second.ToURI());
1775
1776   dict->SetInteger("active_streams", active_streams_.size());
1777
1778   dict->SetInteger("unclaimed_pushed_streams",
1779                    unclaimed_pushed_streams_.size());
1780
1781   dict->SetBoolean("is_secure", is_secure_);
1782
1783   dict->SetString("protocol_negotiated",
1784                   SSLClientSocket::NextProtoToString(
1785                       connection_->socket()->GetNegotiatedProtocol()));
1786
1787   dict->SetInteger("error", error_on_close_);
1788   dict->SetInteger("max_concurrent_streams", max_concurrent_streams_);
1789
1790   dict->SetInteger("streams_initiated_count", streams_initiated_count_);
1791   dict->SetInteger("streams_pushed_count", streams_pushed_count_);
1792   dict->SetInteger("streams_pushed_and_claimed_count",
1793       streams_pushed_and_claimed_count_);
1794   dict->SetInteger("streams_abandoned_count", streams_abandoned_count_);
1795   DCHECK(buffered_spdy_framer_.get());
1796   dict->SetInteger("frames_received", buffered_spdy_framer_->frames_received());
1797
1798   dict->SetBoolean("sent_settings", sent_settings_);
1799   dict->SetBoolean("received_settings", received_settings_);
1800
1801   dict->SetInteger("send_window_size", session_send_window_size_);
1802   dict->SetInteger("recv_window_size", session_recv_window_size_);
1803   dict->SetInteger("unacked_recv_window_bytes",
1804                    session_unacked_recv_window_bytes_);
1805   return dict;
1806 }
1807
1808 bool SpdySession::IsReused() const {
1809   return buffered_spdy_framer_->frames_received() > 0 ||
1810       connection_->reuse_type() == ClientSocketHandle::UNUSED_IDLE;
1811 }
1812
1813 bool SpdySession::GetLoadTimingInfo(SpdyStreamId stream_id,
1814                                     LoadTimingInfo* load_timing_info) const {
1815   return connection_->GetLoadTimingInfo(stream_id != kFirstStreamId,
1816                                         load_timing_info);
1817 }
1818
1819 int SpdySession::GetPeerAddress(IPEndPoint* address) const {
1820   int rv = ERR_SOCKET_NOT_CONNECTED;
1821   if (connection_->socket()) {
1822     rv = connection_->socket()->GetPeerAddress(address);
1823   }
1824
1825   UMA_HISTOGRAM_BOOLEAN("Net.SpdySessionSocketNotConnectedGetPeerAddress",
1826                         rv == ERR_SOCKET_NOT_CONNECTED);
1827
1828   return rv;
1829 }
1830
1831 int SpdySession::GetLocalAddress(IPEndPoint* address) const {
1832   int rv = ERR_SOCKET_NOT_CONNECTED;
1833   if (connection_->socket()) {
1834     rv = connection_->socket()->GetLocalAddress(address);
1835   }
1836
1837   UMA_HISTOGRAM_BOOLEAN("Net.SpdySessionSocketNotConnectedGetLocalAddress",
1838                         rv == ERR_SOCKET_NOT_CONNECTED);
1839
1840   return rv;
1841 }
1842
1843 void SpdySession::EnqueueSessionWrite(RequestPriority priority,
1844                                       SpdyFrameType frame_type,
1845                                       scoped_ptr<SpdyFrame> frame) {
1846   DCHECK(frame_type == RST_STREAM || frame_type == SETTINGS ||
1847          frame_type == WINDOW_UPDATE || frame_type == PING ||
1848          frame_type == GOAWAY);
1849   EnqueueWrite(
1850       priority, frame_type,
1851       scoped_ptr<SpdyBufferProducer>(
1852           new SimpleBufferProducer(
1853               scoped_ptr<SpdyBuffer>(new SpdyBuffer(frame.Pass())))),
1854       base::WeakPtr<SpdyStream>());
1855 }
1856
1857 void SpdySession::EnqueueWrite(RequestPriority priority,
1858                                SpdyFrameType frame_type,
1859                                scoped_ptr<SpdyBufferProducer> producer,
1860                                const base::WeakPtr<SpdyStream>& stream) {
1861   if (availability_state_ == STATE_DRAINING)
1862     return;
1863
1864   write_queue_.Enqueue(priority, frame_type, producer.Pass(), stream);
1865   MaybePostWriteLoop();
1866 }
1867
1868 void SpdySession::MaybePostWriteLoop() {
1869   if (write_state_ == WRITE_STATE_IDLE) {
1870     CHECK(!in_flight_write_);
1871     write_state_ = WRITE_STATE_DO_WRITE;
1872     base::MessageLoop::current()->PostTask(
1873         FROM_HERE,
1874         base::Bind(&SpdySession::PumpWriteLoop,
1875                    weak_factory_.GetWeakPtr(), WRITE_STATE_DO_WRITE, OK));
1876   }
1877 }
1878
1879 void SpdySession::InsertCreatedStream(scoped_ptr<SpdyStream> stream) {
1880   CHECK_EQ(stream->stream_id(), 0u);
1881   CHECK(created_streams_.find(stream.get()) == created_streams_.end());
1882   created_streams_.insert(stream.release());
1883 }
1884
1885 scoped_ptr<SpdyStream> SpdySession::ActivateCreatedStream(SpdyStream* stream) {
1886   CHECK_EQ(stream->stream_id(), 0u);
1887   CHECK(created_streams_.find(stream) != created_streams_.end());
1888   stream->set_stream_id(GetNewStreamId());
1889   scoped_ptr<SpdyStream> owned_stream(stream);
1890   created_streams_.erase(stream);
1891   return owned_stream.Pass();
1892 }
1893
1894 void SpdySession::InsertActivatedStream(scoped_ptr<SpdyStream> stream) {
1895   SpdyStreamId stream_id = stream->stream_id();
1896   CHECK_NE(stream_id, 0u);
1897   std::pair<ActiveStreamMap::iterator, bool> result =
1898       active_streams_.insert(
1899           std::make_pair(stream_id, ActiveStreamInfo(stream.get())));
1900   CHECK(result.second);
1901   ignore_result(stream.release());
1902 }
1903
1904 void SpdySession::DeleteStream(scoped_ptr<SpdyStream> stream, int status) {
1905   if (in_flight_write_stream_.get() == stream.get()) {
1906     // If we're deleting the stream for the in-flight write, we still
1907     // need to let the write complete, so we clear
1908     // |in_flight_write_stream_| and let the write finish on its own
1909     // without notifying |in_flight_write_stream_|.
1910     in_flight_write_stream_.reset();
1911   }
1912
1913   write_queue_.RemovePendingWritesForStream(stream->GetWeakPtr());
1914   stream->OnClose(status);
1915
1916   if (availability_state_ == STATE_AVAILABLE) {
1917     ProcessPendingStreamRequests();
1918   }
1919 }
1920
1921 base::WeakPtr<SpdyStream> SpdySession::GetActivePushStream(const GURL& url) {
1922   base::StatsCounter used_push_streams("spdy.claimed_push_streams");
1923
1924   PushedStreamMap::iterator unclaimed_it = unclaimed_pushed_streams_.find(url);
1925   if (unclaimed_it == unclaimed_pushed_streams_.end())
1926     return base::WeakPtr<SpdyStream>();
1927
1928   SpdyStreamId stream_id = unclaimed_it->second.stream_id;
1929   unclaimed_pushed_streams_.erase(unclaimed_it);
1930
1931   ActiveStreamMap::iterator active_it = active_streams_.find(stream_id);
1932   if (active_it == active_streams_.end()) {
1933     NOTREACHED();
1934     return base::WeakPtr<SpdyStream>();
1935   }
1936
1937   net_log_.AddEvent(NetLog::TYPE_SPDY_STREAM_ADOPTED_PUSH_STREAM);
1938   used_push_streams.Increment();
1939   return active_it->second.stream->GetWeakPtr();
1940 }
1941
1942 bool SpdySession::GetSSLInfo(SSLInfo* ssl_info,
1943                              bool* was_npn_negotiated,
1944                              NextProto* protocol_negotiated) {
1945   *was_npn_negotiated = connection_->socket()->WasNpnNegotiated();
1946   *protocol_negotiated = connection_->socket()->GetNegotiatedProtocol();
1947   return connection_->socket()->GetSSLInfo(ssl_info);
1948 }
1949
1950 bool SpdySession::GetSSLCertRequestInfo(
1951     SSLCertRequestInfo* cert_request_info) {
1952   if (!is_secure_)
1953     return false;
1954   GetSSLClientSocket()->GetSSLCertRequestInfo(cert_request_info);
1955   return true;
1956 }
1957
1958 void SpdySession::OnError(SpdyFramer::SpdyError error_code) {
1959   CHECK(in_io_loop_);
1960
1961   RecordProtocolErrorHistogram(MapFramerErrorToProtocolError(error_code));
1962   std::string description =
1963       base::StringPrintf("Framer error: %d (%s).",
1964                          error_code,
1965                          SpdyFramer::ErrorCodeToString(error_code));
1966   DoDrainSession(MapFramerErrorToNetError(error_code), description);
1967 }
1968
1969 void SpdySession::OnStreamError(SpdyStreamId stream_id,
1970                                 const std::string& description) {
1971   CHECK(in_io_loop_);
1972
1973   ActiveStreamMap::iterator it = active_streams_.find(stream_id);
1974   if (it == active_streams_.end()) {
1975     // We still want to send a frame to reset the stream even if we
1976     // don't know anything about it.
1977     EnqueueResetStreamFrame(
1978         stream_id, IDLE, RST_STREAM_PROTOCOL_ERROR, description);
1979     return;
1980   }
1981
1982   ResetStreamIterator(it, RST_STREAM_PROTOCOL_ERROR, description);
1983 }
1984
1985 void SpdySession::OnDataFrameHeader(SpdyStreamId stream_id,
1986                                     size_t length,
1987                                     bool fin) {
1988   CHECK(in_io_loop_);
1989
1990   ActiveStreamMap::iterator it = active_streams_.find(stream_id);
1991
1992   // By the time data comes in, the stream may already be inactive.
1993   if (it == active_streams_.end())
1994     return;
1995
1996   SpdyStream* stream = it->second.stream;
1997   CHECK_EQ(stream->stream_id(), stream_id);
1998
1999   DCHECK(buffered_spdy_framer_);
2000   size_t header_len = buffered_spdy_framer_->GetDataFrameMinimumSize();
2001   stream->IncrementRawReceivedBytes(header_len);
2002 }
2003
2004 void SpdySession::OnStreamFrameData(SpdyStreamId stream_id,
2005                                     const char* data,
2006                                     size_t len,
2007                                     bool fin) {
2008   CHECK(in_io_loop_);
2009
2010   if (data == NULL && len != 0) {
2011     // This is notification of consumed data padding.
2012     // TODO(jgraettinger): Properly flow padding into WINDOW_UPDATE frames.
2013     // See crbug.com/353012.
2014     return;
2015   }
2016
2017   DCHECK_LT(len, 1u << 24);
2018   if (net_log().IsLogging()) {
2019     net_log().AddEvent(
2020         NetLog::TYPE_SPDY_SESSION_RECV_DATA,
2021         base::Bind(&NetLogSpdyDataCallback, stream_id, len, fin));
2022   }
2023
2024   // Build the buffer as early as possible so that we go through the
2025   // session flow control checks and update
2026   // |unacked_recv_window_bytes_| properly even when the stream is
2027   // inactive (since the other side has still reduced its session send
2028   // window).
2029   scoped_ptr<SpdyBuffer> buffer;
2030   if (data) {
2031     DCHECK_GT(len, 0u);
2032     CHECK_LE(len, static_cast<size_t>(kReadBufferSize));
2033     buffer.reset(new SpdyBuffer(data, len));
2034
2035     if (flow_control_state_ == FLOW_CONTROL_STREAM_AND_SESSION) {
2036       DecreaseRecvWindowSize(static_cast<int32>(len));
2037       buffer->AddConsumeCallback(
2038           base::Bind(&SpdySession::OnReadBufferConsumed,
2039                      weak_factory_.GetWeakPtr()));
2040     }
2041   } else {
2042     DCHECK_EQ(len, 0u);
2043   }
2044
2045   ActiveStreamMap::iterator it = active_streams_.find(stream_id);
2046
2047   // By the time data comes in, the stream may already be inactive.
2048   if (it == active_streams_.end())
2049     return;
2050
2051   SpdyStream* stream = it->second.stream;
2052   CHECK_EQ(stream->stream_id(), stream_id);
2053
2054   stream->IncrementRawReceivedBytes(len);
2055
2056   if (it->second.waiting_for_syn_reply) {
2057     const std::string& error = "Data received before SYN_REPLY.";
2058     stream->LogStreamError(ERR_SPDY_PROTOCOL_ERROR, error);
2059     ResetStreamIterator(it, RST_STREAM_PROTOCOL_ERROR, error);
2060     return;
2061   }
2062
2063   stream->OnDataReceived(buffer.Pass());
2064 }
2065
2066 void SpdySession::OnSettings(bool clear_persisted) {
2067   CHECK(in_io_loop_);
2068
2069   if (clear_persisted)
2070     http_server_properties_->ClearSpdySettings(host_port_pair());
2071
2072   if (net_log_.IsLogging()) {
2073     net_log_.AddEvent(
2074         NetLog::TYPE_SPDY_SESSION_RECV_SETTINGS,
2075         base::Bind(&NetLogSpdySettingsCallback, host_port_pair(),
2076                    clear_persisted));
2077   }
2078
2079   if (GetProtocolVersion() >= SPDY4) {
2080     // Send an acknowledgment of the setting.
2081     SpdySettingsIR settings_ir;
2082     settings_ir.set_is_ack(true);
2083     EnqueueSessionWrite(
2084         HIGHEST,
2085         SETTINGS,
2086         scoped_ptr<SpdyFrame>(
2087             buffered_spdy_framer_->SerializeFrame(settings_ir)));
2088   }
2089 }
2090
2091 void SpdySession::OnSetting(SpdySettingsIds id,
2092                             uint8 flags,
2093                             uint32 value) {
2094   CHECK(in_io_loop_);
2095
2096   HandleSetting(id, value);
2097   http_server_properties_->SetSpdySetting(
2098       host_port_pair(),
2099       id,
2100       static_cast<SpdySettingsFlags>(flags),
2101       value);
2102   received_settings_ = true;
2103
2104   // Log the setting.
2105   net_log_.AddEvent(
2106       NetLog::TYPE_SPDY_SESSION_RECV_SETTING,
2107       base::Bind(&NetLogSpdySettingCallback,
2108                  id, static_cast<SpdySettingsFlags>(flags), value));
2109 }
2110
2111 void SpdySession::OnSendCompressedFrame(
2112     SpdyStreamId stream_id,
2113     SpdyFrameType type,
2114     size_t payload_len,
2115     size_t frame_len) {
2116   if (type != SYN_STREAM && type != HEADERS)
2117     return;
2118
2119   DCHECK(buffered_spdy_framer_.get());
2120   size_t compressed_len =
2121       frame_len - buffered_spdy_framer_->GetSynStreamMinimumSize();
2122
2123   if (payload_len) {
2124     // Make sure we avoid early decimal truncation.
2125     int compression_pct = 100 - (100 * compressed_len) / payload_len;
2126     UMA_HISTOGRAM_PERCENTAGE("Net.SpdySynStreamCompressionPercentage",
2127                              compression_pct);
2128   }
2129 }
2130
2131 void SpdySession::OnReceiveCompressedFrame(
2132     SpdyStreamId stream_id,
2133     SpdyFrameType type,
2134     size_t frame_len) {
2135   last_compressed_frame_len_ = frame_len;
2136 }
2137
2138 int SpdySession::OnInitialResponseHeadersReceived(
2139     const SpdyHeaderBlock& response_headers,
2140     base::Time response_time,
2141     base::TimeTicks recv_first_byte_time,
2142     SpdyStream* stream) {
2143   CHECK(in_io_loop_);
2144   SpdyStreamId stream_id = stream->stream_id();
2145
2146   if (stream->type() == SPDY_PUSH_STREAM) {
2147     DCHECK(stream->IsReservedRemote());
2148     if (max_concurrent_pushed_streams_ &&
2149         num_active_pushed_streams_ >= max_concurrent_pushed_streams_) {
2150       ResetStream(stream_id,
2151                   RST_STREAM_REFUSED_STREAM,
2152                   "Stream concurrency limit reached.");
2153       return STATUS_CODE_REFUSED_STREAM;
2154     }
2155   }
2156
2157   if (stream->type() == SPDY_PUSH_STREAM) {
2158     // Will be balanced in DeleteStream.
2159     num_active_pushed_streams_++;
2160   }
2161
2162   // May invalidate |stream|.
2163   int rv = stream->OnInitialResponseHeadersReceived(
2164       response_headers, response_time, recv_first_byte_time);
2165   if (rv < 0) {
2166     DCHECK_NE(rv, ERR_IO_PENDING);
2167     DCHECK(active_streams_.find(stream_id) == active_streams_.end());
2168   }
2169
2170   return rv;
2171 }
2172
2173 void SpdySession::OnSynStream(SpdyStreamId stream_id,
2174                               SpdyStreamId associated_stream_id,
2175                               SpdyPriority priority,
2176                               bool fin,
2177                               bool unidirectional,
2178                               const SpdyHeaderBlock& headers) {
2179   CHECK(in_io_loop_);
2180
2181   if (GetProtocolVersion() >= SPDY4) {
2182     DCHECK_EQ(0u, associated_stream_id);
2183     OnHeaders(stream_id, fin, headers);
2184     return;
2185   }
2186
2187   base::Time response_time = base::Time::Now();
2188   base::TimeTicks recv_first_byte_time = time_func_();
2189
2190   if (net_log_.IsLogging()) {
2191     net_log_.AddEvent(
2192         NetLog::TYPE_SPDY_SESSION_PUSHED_SYN_STREAM,
2193         base::Bind(&NetLogSpdySynStreamReceivedCallback,
2194                    &headers, fin, unidirectional, priority,
2195                    stream_id, associated_stream_id));
2196   }
2197
2198   // Split headers to simulate push promise and response.
2199   SpdyHeaderBlock request_headers;
2200   SpdyHeaderBlock response_headers;
2201   SplitPushedHeadersToRequestAndResponse(
2202       headers, GetProtocolVersion(), &request_headers, &response_headers);
2203
2204   if (!TryCreatePushStream(
2205           stream_id, associated_stream_id, priority, request_headers))
2206     return;
2207
2208   ActiveStreamMap::iterator active_it = active_streams_.find(stream_id);
2209   if (active_it == active_streams_.end()) {
2210     NOTREACHED();
2211     return;
2212   }
2213
2214   if (OnInitialResponseHeadersReceived(response_headers,
2215                                        response_time,
2216                                        recv_first_byte_time,
2217                                        active_it->second.stream) != OK)
2218     return;
2219
2220   base::StatsCounter push_requests("spdy.pushed_streams");
2221   push_requests.Increment();
2222 }
2223
2224 void SpdySession::DeleteExpiredPushedStreams() {
2225   if (unclaimed_pushed_streams_.empty())
2226     return;
2227
2228   // Check that adequate time has elapsed since the last sweep.
2229   if (time_func_() < next_unclaimed_push_stream_sweep_time_)
2230     return;
2231
2232   // Gather old streams to delete.
2233   base::TimeTicks minimum_freshness = time_func_() -
2234       base::TimeDelta::FromSeconds(kMinPushedStreamLifetimeSeconds);
2235   std::vector<SpdyStreamId> streams_to_close;
2236   for (PushedStreamMap::iterator it = unclaimed_pushed_streams_.begin();
2237        it != unclaimed_pushed_streams_.end(); ++it) {
2238     if (minimum_freshness > it->second.creation_time)
2239       streams_to_close.push_back(it->second.stream_id);
2240   }
2241
2242   for (std::vector<SpdyStreamId>::const_iterator to_close_it =
2243            streams_to_close.begin();
2244        to_close_it != streams_to_close.end(); ++to_close_it) {
2245     ActiveStreamMap::iterator active_it = active_streams_.find(*to_close_it);
2246     if (active_it == active_streams_.end())
2247       continue;
2248
2249     LogAbandonedActiveStream(active_it, ERR_INVALID_SPDY_STREAM);
2250     // CloseActiveStreamIterator() will remove the stream from
2251     // |unclaimed_pushed_streams_|.
2252     ResetStreamIterator(
2253         active_it, RST_STREAM_REFUSED_STREAM, "Stream not claimed.");
2254   }
2255
2256   next_unclaimed_push_stream_sweep_time_ = time_func_() +
2257       base::TimeDelta::FromSeconds(kMinPushedStreamLifetimeSeconds);
2258 }
2259
2260 void SpdySession::OnSynReply(SpdyStreamId stream_id,
2261                              bool fin,
2262                              const SpdyHeaderBlock& headers) {
2263   CHECK(in_io_loop_);
2264
2265   base::Time response_time = base::Time::Now();
2266   base::TimeTicks recv_first_byte_time = time_func_();
2267
2268   if (net_log().IsLogging()) {
2269     net_log().AddEvent(
2270         NetLog::TYPE_SPDY_SESSION_SYN_REPLY,
2271         base::Bind(&NetLogSpdySynReplyOrHeadersReceivedCallback,
2272                    &headers, fin, stream_id));
2273   }
2274
2275   ActiveStreamMap::iterator it = active_streams_.find(stream_id);
2276   if (it == active_streams_.end()) {
2277     // NOTE:  it may just be that the stream was cancelled.
2278     return;
2279   }
2280
2281   SpdyStream* stream = it->second.stream;
2282   CHECK_EQ(stream->stream_id(), stream_id);
2283
2284   stream->IncrementRawReceivedBytes(last_compressed_frame_len_);
2285   last_compressed_frame_len_ = 0;
2286
2287   if (GetProtocolVersion() >= SPDY4) {
2288     const std::string& error =
2289         "SPDY4 wasn't expecting SYN_REPLY.";
2290     stream->LogStreamError(ERR_SPDY_PROTOCOL_ERROR, error);
2291     ResetStreamIterator(it, RST_STREAM_PROTOCOL_ERROR, error);
2292     return;
2293   }
2294   if (!it->second.waiting_for_syn_reply) {
2295     const std::string& error =
2296         "Received duplicate SYN_REPLY for stream.";
2297     stream->LogStreamError(ERR_SPDY_PROTOCOL_ERROR, error);
2298     ResetStreamIterator(it, RST_STREAM_PROTOCOL_ERROR, error);
2299     return;
2300   }
2301   it->second.waiting_for_syn_reply = false;
2302
2303   ignore_result(OnInitialResponseHeadersReceived(
2304       headers, response_time, recv_first_byte_time, stream));
2305 }
2306
2307 void SpdySession::OnHeaders(SpdyStreamId stream_id,
2308                             bool fin,
2309                             const SpdyHeaderBlock& headers) {
2310   CHECK(in_io_loop_);
2311
2312   if (net_log().IsLogging()) {
2313     net_log().AddEvent(
2314         NetLog::TYPE_SPDY_SESSION_RECV_HEADERS,
2315         base::Bind(&NetLogSpdySynReplyOrHeadersReceivedCallback,
2316                    &headers, fin, stream_id));
2317   }
2318
2319   ActiveStreamMap::iterator it = active_streams_.find(stream_id);
2320   if (it == active_streams_.end()) {
2321     // NOTE:  it may just be that the stream was cancelled.
2322     LOG(WARNING) << "Received HEADERS for invalid stream " << stream_id;
2323     return;
2324   }
2325
2326   SpdyStream* stream = it->second.stream;
2327   CHECK_EQ(stream->stream_id(), stream_id);
2328
2329   stream->IncrementRawReceivedBytes(last_compressed_frame_len_);
2330   last_compressed_frame_len_ = 0;
2331
2332   base::Time response_time = base::Time::Now();
2333   base::TimeTicks recv_first_byte_time = time_func_();
2334
2335   if (it->second.waiting_for_syn_reply) {
2336     if (GetProtocolVersion() < SPDY4) {
2337       const std::string& error =
2338           "Was expecting SYN_REPLY, not HEADERS.";
2339       stream->LogStreamError(ERR_SPDY_PROTOCOL_ERROR, error);
2340       ResetStreamIterator(it, RST_STREAM_PROTOCOL_ERROR, error);
2341       return;
2342     }
2343
2344     it->second.waiting_for_syn_reply = false;
2345     ignore_result(OnInitialResponseHeadersReceived(
2346         headers, response_time, recv_first_byte_time, stream));
2347   } else if (it->second.stream->IsReservedRemote()) {
2348     ignore_result(OnInitialResponseHeadersReceived(
2349         headers, response_time, recv_first_byte_time, stream));
2350   } else {
2351     int rv = stream->OnAdditionalResponseHeadersReceived(headers);
2352     if (rv < 0) {
2353       DCHECK_NE(rv, ERR_IO_PENDING);
2354       DCHECK(active_streams_.find(stream_id) == active_streams_.end());
2355     }
2356   }
2357 }
2358
2359 bool SpdySession::OnUnknownFrame(SpdyStreamId stream_id, int frame_type) {
2360   // Validate stream id.
2361   // Was the frame sent on a stream id that has not been used in this session?
2362   if (stream_id % 2 == 1 && stream_id > stream_hi_water_mark_)
2363     return false;
2364
2365   if (stream_id % 2 == 0 && stream_id > last_accepted_push_stream_id_)
2366     return false;
2367
2368   return true;
2369 }
2370
2371 void SpdySession::OnRstStream(SpdyStreamId stream_id,
2372                               SpdyRstStreamStatus status) {
2373   CHECK(in_io_loop_);
2374
2375   std::string description;
2376   net_log().AddEvent(
2377       NetLog::TYPE_SPDY_SESSION_RST_STREAM,
2378       base::Bind(&NetLogSpdyRstCallback,
2379                  stream_id, status, &description));
2380
2381   ActiveStreamMap::iterator it = active_streams_.find(stream_id);
2382   if (it == active_streams_.end()) {
2383     // NOTE:  it may just be that the stream was cancelled.
2384     LOG(WARNING) << "Received RST for invalid stream" << stream_id;
2385     return;
2386   }
2387
2388   CHECK_EQ(it->second.stream->stream_id(), stream_id);
2389
2390   if (status == 0) {
2391     it->second.stream->OnDataReceived(scoped_ptr<SpdyBuffer>());
2392   } else if (status == RST_STREAM_REFUSED_STREAM) {
2393     CloseActiveStreamIterator(it, ERR_SPDY_SERVER_REFUSED_STREAM);
2394   } else {
2395     RecordProtocolErrorHistogram(
2396         PROTOCOL_ERROR_RST_STREAM_FOR_NON_ACTIVE_STREAM);
2397     it->second.stream->LogStreamError(
2398         ERR_SPDY_PROTOCOL_ERROR,
2399         base::StringPrintf("SPDY stream closed with status: %d", status));
2400     // TODO(mbelshe): Map from Spdy-protocol errors to something sensical.
2401     //                For now, it doesn't matter much - it is a protocol error.
2402     CloseActiveStreamIterator(it, ERR_SPDY_PROTOCOL_ERROR);
2403   }
2404 }
2405
2406 void SpdySession::OnGoAway(SpdyStreamId last_accepted_stream_id,
2407                            SpdyGoAwayStatus status) {
2408   CHECK(in_io_loop_);
2409
2410   // TODO(jgraettinger): UMA histogram on |status|.
2411
2412   net_log_.AddEvent(NetLog::TYPE_SPDY_SESSION_GOAWAY,
2413       base::Bind(&NetLogSpdyGoAwayCallback,
2414                  last_accepted_stream_id,
2415                  active_streams_.size(),
2416                  unclaimed_pushed_streams_.size(),
2417                  status));
2418   MakeUnavailable();
2419   StartGoingAway(last_accepted_stream_id, ERR_ABORTED);
2420   // This is to handle the case when we already don't have any active
2421   // streams (i.e., StartGoingAway() did nothing). Otherwise, we have
2422   // active streams and so the last one being closed will finish the
2423   // going away process (see DeleteStream()).
2424   MaybeFinishGoingAway();
2425 }
2426
2427 void SpdySession::OnPing(SpdyPingId unique_id, bool is_ack) {
2428   CHECK(in_io_loop_);
2429
2430   net_log_.AddEvent(
2431       NetLog::TYPE_SPDY_SESSION_PING,
2432       base::Bind(&NetLogSpdyPingCallback, unique_id, is_ack, "received"));
2433
2434   // Send response to a PING from server.
2435   if ((protocol_ >= kProtoSPDY4 && !is_ack) ||
2436       (protocol_ < kProtoSPDY4 && unique_id % 2 == 0)) {
2437     WritePingFrame(unique_id, true);
2438     return;
2439   }
2440
2441   --pings_in_flight_;
2442   if (pings_in_flight_ < 0) {
2443     RecordProtocolErrorHistogram(PROTOCOL_ERROR_UNEXPECTED_PING);
2444     DoDrainSession(ERR_SPDY_PROTOCOL_ERROR, "pings_in_flight_ is < 0.");
2445     pings_in_flight_ = 0;
2446     return;
2447   }
2448
2449   if (pings_in_flight_ > 0)
2450     return;
2451
2452   // We will record RTT in histogram when there are no more client sent
2453   // pings_in_flight_.
2454   RecordPingRTTHistogram(time_func_() - last_ping_sent_time_);
2455 }
2456
2457 void SpdySession::OnWindowUpdate(SpdyStreamId stream_id,
2458                                  uint32 delta_window_size) {
2459   CHECK(in_io_loop_);
2460
2461   DCHECK_LE(delta_window_size, static_cast<uint32>(kint32max));
2462   net_log_.AddEvent(
2463       NetLog::TYPE_SPDY_SESSION_RECEIVED_WINDOW_UPDATE_FRAME,
2464       base::Bind(&NetLogSpdyWindowUpdateFrameCallback,
2465                  stream_id, delta_window_size));
2466
2467   if (stream_id == kSessionFlowControlStreamId) {
2468     // WINDOW_UPDATE for the session.
2469     if (flow_control_state_ < FLOW_CONTROL_STREAM_AND_SESSION) {
2470       LOG(WARNING) << "Received WINDOW_UPDATE for session when "
2471                    << "session flow control is not turned on";
2472       // TODO(akalin): Record an error and close the session.
2473       return;
2474     }
2475
2476     if (delta_window_size < 1u) {
2477       RecordProtocolErrorHistogram(PROTOCOL_ERROR_INVALID_WINDOW_UPDATE_SIZE);
2478       DoDrainSession(
2479           ERR_SPDY_PROTOCOL_ERROR,
2480           "Received WINDOW_UPDATE with an invalid delta_window_size " +
2481               base::UintToString(delta_window_size));
2482       return;
2483     }
2484
2485     IncreaseSendWindowSize(static_cast<int32>(delta_window_size));
2486   } else {
2487     // WINDOW_UPDATE for a stream.
2488     if (flow_control_state_ < FLOW_CONTROL_STREAM) {
2489       // TODO(akalin): Record an error and close the session.
2490       LOG(WARNING) << "Received WINDOW_UPDATE for stream " << stream_id
2491                    << " when flow control is not turned on";
2492       return;
2493     }
2494
2495     ActiveStreamMap::iterator it = active_streams_.find(stream_id);
2496
2497     if (it == active_streams_.end()) {
2498       // NOTE:  it may just be that the stream was cancelled.
2499       LOG(WARNING) << "Received WINDOW_UPDATE for invalid stream " << stream_id;
2500       return;
2501     }
2502
2503     SpdyStream* stream = it->second.stream;
2504     CHECK_EQ(stream->stream_id(), stream_id);
2505
2506     if (delta_window_size < 1u) {
2507       ResetStreamIterator(it,
2508                           RST_STREAM_FLOW_CONTROL_ERROR,
2509                           base::StringPrintf(
2510                               "Received WINDOW_UPDATE with an invalid "
2511                               "delta_window_size %ud", delta_window_size));
2512       return;
2513     }
2514
2515     CHECK_EQ(it->second.stream->stream_id(), stream_id);
2516     it->second.stream->IncreaseSendWindowSize(
2517         static_cast<int32>(delta_window_size));
2518   }
2519 }
2520
2521 bool SpdySession::TryCreatePushStream(SpdyStreamId stream_id,
2522                                       SpdyStreamId associated_stream_id,
2523                                       SpdyPriority priority,
2524                                       const SpdyHeaderBlock& headers) {
2525   // Server-initiated streams should have even sequence numbers.
2526   if ((stream_id & 0x1) != 0) {
2527     LOG(WARNING) << "Received invalid push stream id " << stream_id;
2528     if (GetProtocolVersion() > SPDY2)
2529       CloseSessionOnError(ERR_SPDY_PROTOCOL_ERROR, "Odd push stream id.");
2530     return false;
2531   }
2532
2533   if (GetProtocolVersion() > SPDY2) {
2534     if (stream_id <= last_accepted_push_stream_id_) {
2535       LOG(WARNING) << "Received push stream id lesser or equal to the last "
2536                    << "accepted before " << stream_id;
2537       CloseSessionOnError(
2538           ERR_SPDY_PROTOCOL_ERROR,
2539           "New push stream id must be greater than the last accepted.");
2540       return false;
2541     }
2542   }
2543
2544   if (IsStreamActive(stream_id)) {
2545     // For SPDY3 and higher we should not get here, we'll start going away
2546     // earlier on |last_seen_push_stream_id_| check.
2547     CHECK_GT(SPDY3, GetProtocolVersion());
2548     LOG(WARNING) << "Received push for active stream " << stream_id;
2549     return false;
2550   }
2551
2552   last_accepted_push_stream_id_ = stream_id;
2553
2554   RequestPriority request_priority =
2555       ConvertSpdyPriorityToRequestPriority(priority, GetProtocolVersion());
2556
2557   if (availability_state_ == STATE_GOING_AWAY) {
2558     // TODO(akalin): This behavior isn't in the SPDY spec, although it
2559     // probably should be.
2560     EnqueueResetStreamFrame(stream_id,
2561                             request_priority,
2562                             RST_STREAM_REFUSED_STREAM,
2563                             "push stream request received when going away");
2564     return false;
2565   }
2566
2567   if (associated_stream_id == 0) {
2568     // In SPDY4 0 stream id in PUSH_PROMISE frame leads to framer error and
2569     // session going away. We should never get here.
2570     CHECK_GT(SPDY4, GetProtocolVersion());
2571     std::string description = base::StringPrintf(
2572         "Received invalid associated stream id %d for pushed stream %d",
2573         associated_stream_id,
2574         stream_id);
2575     EnqueueResetStreamFrame(
2576         stream_id, request_priority, RST_STREAM_REFUSED_STREAM, description);
2577     return false;
2578   }
2579
2580   streams_pushed_count_++;
2581
2582   // TODO(mbelshe): DCHECK that this is a GET method?
2583
2584   // Verify that the response had a URL for us.
2585   GURL gurl = GetUrlFromHeaderBlock(headers, GetProtocolVersion(), true);
2586   if (!gurl.is_valid()) {
2587     EnqueueResetStreamFrame(stream_id,
2588                             request_priority,
2589                             RST_STREAM_PROTOCOL_ERROR,
2590                             "Pushed stream url was invalid: " + gurl.spec());
2591     return false;
2592   }
2593
2594   // Verify we have a valid stream association.
2595   ActiveStreamMap::iterator associated_it =
2596       active_streams_.find(associated_stream_id);
2597   if (associated_it == active_streams_.end()) {
2598     EnqueueResetStreamFrame(
2599         stream_id,
2600         request_priority,
2601         RST_STREAM_INVALID_STREAM,
2602         base::StringPrintf("Received push for inactive associated stream %d",
2603                            associated_stream_id));
2604     return false;
2605   }
2606
2607   // Check that the pushed stream advertises the same origin as its associated
2608   // stream. Bypass this check if and only if this session is with a SPDY proxy
2609   // that is trusted explicitly via the --trusted-spdy-proxy switch.
2610   if (trusted_spdy_proxy_.Equals(host_port_pair())) {
2611     // Disallow pushing of HTTPS content.
2612     if (gurl.SchemeIs("https")) {
2613       EnqueueResetStreamFrame(
2614           stream_id,
2615           request_priority,
2616           RST_STREAM_REFUSED_STREAM,
2617           base::StringPrintf("Rejected push of Cross Origin HTTPS content %d",
2618                              associated_stream_id));
2619     }
2620   } else {
2621     GURL associated_url(associated_it->second.stream->GetUrlFromHeaders());
2622     if (associated_url.GetOrigin() != gurl.GetOrigin()) {
2623       EnqueueResetStreamFrame(
2624           stream_id,
2625           request_priority,
2626           RST_STREAM_REFUSED_STREAM,
2627           base::StringPrintf("Rejected Cross Origin Push Stream %d",
2628                              associated_stream_id));
2629       return false;
2630     }
2631   }
2632
2633   // There should not be an existing pushed stream with the same path.
2634   PushedStreamMap::iterator pushed_it =
2635       unclaimed_pushed_streams_.lower_bound(gurl);
2636   if (pushed_it != unclaimed_pushed_streams_.end() &&
2637       pushed_it->first == gurl) {
2638     EnqueueResetStreamFrame(
2639         stream_id,
2640         request_priority,
2641         RST_STREAM_PROTOCOL_ERROR,
2642         "Received duplicate pushed stream with url: " + gurl.spec());
2643     return false;
2644   }
2645
2646   scoped_ptr<SpdyStream> stream(new SpdyStream(SPDY_PUSH_STREAM,
2647                                                GetWeakPtr(),
2648                                                gurl,
2649                                                request_priority,
2650                                                stream_initial_send_window_size_,
2651                                                stream_initial_recv_window_size_,
2652                                                net_log_));
2653   stream->set_stream_id(stream_id);
2654
2655   // In spdy4/http2 PUSH_PROMISE arrives on associated stream.
2656   if (associated_it != active_streams_.end() && GetProtocolVersion() >= SPDY4) {
2657     associated_it->second.stream->IncrementRawReceivedBytes(
2658         last_compressed_frame_len_);
2659   } else {
2660     stream->IncrementRawReceivedBytes(last_compressed_frame_len_);
2661   }
2662
2663   last_compressed_frame_len_ = 0;
2664
2665   DeleteExpiredPushedStreams();
2666   PushedStreamMap::iterator inserted_pushed_it =
2667       unclaimed_pushed_streams_.insert(
2668           pushed_it,
2669           std::make_pair(gurl, PushedStreamInfo(stream_id, time_func_())));
2670   DCHECK(inserted_pushed_it != pushed_it);
2671
2672   InsertActivatedStream(stream.Pass());
2673
2674   ActiveStreamMap::iterator active_it = active_streams_.find(stream_id);
2675   if (active_it == active_streams_.end()) {
2676     NOTREACHED();
2677     return false;
2678   }
2679
2680   active_it->second.stream->OnPushPromiseHeadersReceived(headers);
2681   DCHECK(active_it->second.stream->IsReservedRemote());
2682   num_pushed_streams_++;
2683   return true;
2684 }
2685
2686 void SpdySession::OnPushPromise(SpdyStreamId stream_id,
2687                                 SpdyStreamId promised_stream_id,
2688                                 const SpdyHeaderBlock& headers) {
2689   CHECK(in_io_loop_);
2690
2691   if (net_log_.IsLogging()) {
2692     net_log_.AddEvent(NetLog::TYPE_SPDY_SESSION_RECV_PUSH_PROMISE,
2693                       base::Bind(&NetLogSpdyPushPromiseReceivedCallback,
2694                                  &headers,
2695                                  stream_id,
2696                                  promised_stream_id));
2697   }
2698
2699   // Any priority will do.
2700   // TODO(baranovich): pass parent stream id priority?
2701   if (!TryCreatePushStream(promised_stream_id, stream_id, 0, headers))
2702     return;
2703
2704   base::StatsCounter push_requests("spdy.pushed_streams");
2705   push_requests.Increment();
2706 }
2707
2708 void SpdySession::SendStreamWindowUpdate(SpdyStreamId stream_id,
2709                                          uint32 delta_window_size) {
2710   CHECK_GE(flow_control_state_, FLOW_CONTROL_STREAM);
2711   ActiveStreamMap::const_iterator it = active_streams_.find(stream_id);
2712   CHECK(it != active_streams_.end());
2713   CHECK_EQ(it->second.stream->stream_id(), stream_id);
2714   SendWindowUpdateFrame(
2715       stream_id, delta_window_size, it->second.stream->priority());
2716 }
2717
2718 void SpdySession::SendInitialData() {
2719   DCHECK(enable_sending_initial_data_);
2720
2721   if (send_connection_header_prefix_) {
2722     DCHECK_EQ(protocol_, kProtoSPDY4);
2723     scoped_ptr<SpdyFrame> connection_header_prefix_frame(
2724         new SpdyFrame(const_cast<char*>(kHttp2ConnectionHeaderPrefix),
2725                       kHttp2ConnectionHeaderPrefixSize,
2726                       false /* take_ownership */));
2727     // Count the prefix as part of the subsequent SETTINGS frame.
2728     EnqueueSessionWrite(HIGHEST, SETTINGS,
2729                         connection_header_prefix_frame.Pass());
2730   }
2731
2732   // First, notify the server about the settings they should use when
2733   // communicating with us.
2734   SettingsMap settings_map;
2735   // Create a new settings frame notifying the server of our
2736   // max concurrent streams and initial window size.
2737   settings_map[SETTINGS_MAX_CONCURRENT_STREAMS] =
2738       SettingsFlagsAndValue(SETTINGS_FLAG_NONE, kMaxConcurrentPushedStreams);
2739   if (flow_control_state_ >= FLOW_CONTROL_STREAM &&
2740       stream_initial_recv_window_size_ != kSpdyStreamInitialWindowSize) {
2741     settings_map[SETTINGS_INITIAL_WINDOW_SIZE] =
2742         SettingsFlagsAndValue(SETTINGS_FLAG_NONE,
2743                               stream_initial_recv_window_size_);
2744   }
2745   SendSettings(settings_map);
2746
2747   // Next, notify the server about our initial recv window size.
2748   if (flow_control_state_ == FLOW_CONTROL_STREAM_AND_SESSION) {
2749     // Bump up the receive window size to the real initial value. This
2750     // has to go here since the WINDOW_UPDATE frame sent by
2751     // IncreaseRecvWindowSize() call uses |buffered_spdy_framer_|.
2752     DCHECK_GT(kDefaultInitialRecvWindowSize, session_recv_window_size_);
2753     // This condition implies that |kDefaultInitialRecvWindowSize| -
2754     // |session_recv_window_size_| doesn't overflow.
2755     DCHECK_GT(session_recv_window_size_, 0);
2756     IncreaseRecvWindowSize(
2757         kDefaultInitialRecvWindowSize - session_recv_window_size_);
2758   }
2759
2760   if (protocol_ <= kProtoSPDY31) {
2761     // Finally, notify the server about the settings they have
2762     // previously told us to use when communicating with them (after
2763     // applying them).
2764     const SettingsMap& server_settings_map =
2765         http_server_properties_->GetSpdySettings(host_port_pair());
2766     if (server_settings_map.empty())
2767       return;
2768
2769     SettingsMap::const_iterator it =
2770         server_settings_map.find(SETTINGS_CURRENT_CWND);
2771     uint32 cwnd = (it != server_settings_map.end()) ? it->second.second : 0;
2772     UMA_HISTOGRAM_CUSTOM_COUNTS("Net.SpdySettingsCwndSent", cwnd, 1, 200, 100);
2773
2774     for (SettingsMap::const_iterator it = server_settings_map.begin();
2775          it != server_settings_map.end(); ++it) {
2776       const SpdySettingsIds new_id = it->first;
2777       const uint32 new_val = it->second.second;
2778       HandleSetting(new_id, new_val);
2779     }
2780
2781     SendSettings(server_settings_map);
2782   }
2783 }
2784
2785
2786 void SpdySession::SendSettings(const SettingsMap& settings) {
2787   net_log_.AddEvent(
2788       NetLog::TYPE_SPDY_SESSION_SEND_SETTINGS,
2789       base::Bind(&NetLogSpdySendSettingsCallback, &settings));
2790
2791   // Create the SETTINGS frame and send it.
2792   DCHECK(buffered_spdy_framer_.get());
2793   scoped_ptr<SpdyFrame> settings_frame(
2794       buffered_spdy_framer_->CreateSettings(settings));
2795   sent_settings_ = true;
2796   EnqueueSessionWrite(HIGHEST, SETTINGS, settings_frame.Pass());
2797 }
2798
2799 void SpdySession::HandleSetting(uint32 id, uint32 value) {
2800   switch (id) {
2801     case SETTINGS_MAX_CONCURRENT_STREAMS:
2802       max_concurrent_streams_ = std::min(static_cast<size_t>(value),
2803                                          kMaxConcurrentStreamLimit);
2804       ProcessPendingStreamRequests();
2805       break;
2806     case SETTINGS_INITIAL_WINDOW_SIZE: {
2807       if (flow_control_state_ < FLOW_CONTROL_STREAM) {
2808         net_log().AddEvent(
2809             NetLog::TYPE_SPDY_SESSION_INITIAL_WINDOW_SIZE_NO_FLOW_CONTROL);
2810         return;
2811       }
2812
2813       if (value > static_cast<uint32>(kint32max)) {
2814         net_log().AddEvent(
2815             NetLog::TYPE_SPDY_SESSION_INITIAL_WINDOW_SIZE_OUT_OF_RANGE,
2816             NetLog::IntegerCallback("initial_window_size", value));
2817         return;
2818       }
2819
2820       // SETTINGS_INITIAL_WINDOW_SIZE updates initial_send_window_size_ only.
2821       int32 delta_window_size =
2822           static_cast<int32>(value) - stream_initial_send_window_size_;
2823       stream_initial_send_window_size_ = static_cast<int32>(value);
2824       UpdateStreamsSendWindowSize(delta_window_size);
2825       net_log().AddEvent(
2826           NetLog::TYPE_SPDY_SESSION_UPDATE_STREAMS_SEND_WINDOW_SIZE,
2827           NetLog::IntegerCallback("delta_window_size", delta_window_size));
2828       break;
2829     }
2830   }
2831 }
2832
2833 void SpdySession::UpdateStreamsSendWindowSize(int32 delta_window_size) {
2834   DCHECK_GE(flow_control_state_, FLOW_CONTROL_STREAM);
2835   for (ActiveStreamMap::iterator it = active_streams_.begin();
2836        it != active_streams_.end(); ++it) {
2837     it->second.stream->AdjustSendWindowSize(delta_window_size);
2838   }
2839
2840   for (CreatedStreamSet::const_iterator it = created_streams_.begin();
2841        it != created_streams_.end(); it++) {
2842     (*it)->AdjustSendWindowSize(delta_window_size);
2843   }
2844 }
2845
2846 void SpdySession::SendPrefacePingIfNoneInFlight() {
2847   if (pings_in_flight_ || !enable_ping_based_connection_checking_)
2848     return;
2849
2850   base::TimeTicks now = time_func_();
2851   // If there is no activity in the session, then send a preface-PING.
2852   if ((now - last_activity_time_) > connection_at_risk_of_loss_time_)
2853     SendPrefacePing();
2854 }
2855
2856 void SpdySession::SendPrefacePing() {
2857   WritePingFrame(next_ping_id_, false);
2858 }
2859
2860 void SpdySession::SendWindowUpdateFrame(SpdyStreamId stream_id,
2861                                         uint32 delta_window_size,
2862                                         RequestPriority priority) {
2863   CHECK_GE(flow_control_state_, FLOW_CONTROL_STREAM);
2864   ActiveStreamMap::const_iterator it = active_streams_.find(stream_id);
2865   if (it != active_streams_.end()) {
2866     CHECK_EQ(it->second.stream->stream_id(), stream_id);
2867   } else {
2868     CHECK_EQ(flow_control_state_, FLOW_CONTROL_STREAM_AND_SESSION);
2869     CHECK_EQ(stream_id, kSessionFlowControlStreamId);
2870   }
2871
2872   net_log_.AddEvent(
2873       NetLog::TYPE_SPDY_SESSION_SENT_WINDOW_UPDATE_FRAME,
2874       base::Bind(&NetLogSpdyWindowUpdateFrameCallback,
2875                  stream_id, delta_window_size));
2876
2877   DCHECK(buffered_spdy_framer_.get());
2878   scoped_ptr<SpdyFrame> window_update_frame(
2879       buffered_spdy_framer_->CreateWindowUpdate(stream_id, delta_window_size));
2880   EnqueueSessionWrite(priority, WINDOW_UPDATE, window_update_frame.Pass());
2881 }
2882
2883 void SpdySession::WritePingFrame(uint32 unique_id, bool is_ack) {
2884   DCHECK(buffered_spdy_framer_.get());
2885   scoped_ptr<SpdyFrame> ping_frame(
2886       buffered_spdy_framer_->CreatePingFrame(unique_id, is_ack));
2887   EnqueueSessionWrite(HIGHEST, PING, ping_frame.Pass());
2888
2889   if (net_log().IsLogging()) {
2890     net_log().AddEvent(
2891         NetLog::TYPE_SPDY_SESSION_PING,
2892         base::Bind(&NetLogSpdyPingCallback, unique_id, is_ack, "sent"));
2893   }
2894   if (!is_ack) {
2895     next_ping_id_ += 2;
2896     ++pings_in_flight_;
2897     PlanToCheckPingStatus();
2898     last_ping_sent_time_ = time_func_();
2899   }
2900 }
2901
2902 void SpdySession::PlanToCheckPingStatus() {
2903   if (check_ping_status_pending_)
2904     return;
2905
2906   check_ping_status_pending_ = true;
2907   base::MessageLoop::current()->PostDelayedTask(
2908       FROM_HERE,
2909       base::Bind(&SpdySession::CheckPingStatus, weak_factory_.GetWeakPtr(),
2910                  time_func_()), hung_interval_);
2911 }
2912
2913 void SpdySession::CheckPingStatus(base::TimeTicks last_check_time) {
2914   CHECK(!in_io_loop_);
2915
2916   // Check if we got a response back for all PINGs we had sent.
2917   if (pings_in_flight_ == 0) {
2918     check_ping_status_pending_ = false;
2919     return;
2920   }
2921
2922   DCHECK(check_ping_status_pending_);
2923
2924   base::TimeTicks now = time_func_();
2925   base::TimeDelta delay = hung_interval_ - (now - last_activity_time_);
2926
2927   if (delay.InMilliseconds() < 0 || last_activity_time_ < last_check_time) {
2928     // Track all failed PING messages in a separate bucket.
2929     RecordPingRTTHistogram(base::TimeDelta::Max());
2930     DoDrainSession(ERR_SPDY_PING_FAILED, "Failed ping.");
2931     return;
2932   }
2933
2934   // Check the status of connection after a delay.
2935   base::MessageLoop::current()->PostDelayedTask(
2936       FROM_HERE,
2937       base::Bind(&SpdySession::CheckPingStatus, weak_factory_.GetWeakPtr(),
2938                  now),
2939       delay);
2940 }
2941
2942 void SpdySession::RecordPingRTTHistogram(base::TimeDelta duration) {
2943   UMA_HISTOGRAM_TIMES("Net.SpdyPing.RTT", duration);
2944 }
2945
2946 void SpdySession::RecordProtocolErrorHistogram(
2947     SpdyProtocolErrorDetails details) {
2948   UMA_HISTOGRAM_ENUMERATION("Net.SpdySessionErrorDetails2", details,
2949                             NUM_SPDY_PROTOCOL_ERROR_DETAILS);
2950   if (EndsWith(host_port_pair().host(), "google.com", false)) {
2951     UMA_HISTOGRAM_ENUMERATION("Net.SpdySessionErrorDetails_Google2", details,
2952                               NUM_SPDY_PROTOCOL_ERROR_DETAILS);
2953   }
2954 }
2955
2956 void SpdySession::RecordHistograms() {
2957   UMA_HISTOGRAM_CUSTOM_COUNTS("Net.SpdyStreamsPerSession",
2958                               streams_initiated_count_,
2959                               0, 300, 50);
2960   UMA_HISTOGRAM_CUSTOM_COUNTS("Net.SpdyStreamsPushedPerSession",
2961                               streams_pushed_count_,
2962                               0, 300, 50);
2963   UMA_HISTOGRAM_CUSTOM_COUNTS("Net.SpdyStreamsPushedAndClaimedPerSession",
2964                               streams_pushed_and_claimed_count_,
2965                               0, 300, 50);
2966   UMA_HISTOGRAM_CUSTOM_COUNTS("Net.SpdyStreamsAbandonedPerSession",
2967                               streams_abandoned_count_,
2968                               0, 300, 50);
2969   UMA_HISTOGRAM_ENUMERATION("Net.SpdySettingsSent",
2970                             sent_settings_ ? 1 : 0, 2);
2971   UMA_HISTOGRAM_ENUMERATION("Net.SpdySettingsReceived",
2972                             received_settings_ ? 1 : 0, 2);
2973   UMA_HISTOGRAM_CUSTOM_COUNTS("Net.SpdyStreamStallsPerSession",
2974                               stalled_streams_,
2975                               0, 300, 50);
2976   UMA_HISTOGRAM_ENUMERATION("Net.SpdySessionsWithStalls",
2977                             stalled_streams_ > 0 ? 1 : 0, 2);
2978
2979   if (received_settings_) {
2980     // Enumerate the saved settings, and set histograms for it.
2981     const SettingsMap& settings_map =
2982         http_server_properties_->GetSpdySettings(host_port_pair());
2983
2984     SettingsMap::const_iterator it;
2985     for (it = settings_map.begin(); it != settings_map.end(); ++it) {
2986       const SpdySettingsIds id = it->first;
2987       const uint32 val = it->second.second;
2988       switch (id) {
2989         case SETTINGS_CURRENT_CWND:
2990           // Record several different histograms to see if cwnd converges
2991           // for larger volumes of data being sent.
2992           UMA_HISTOGRAM_CUSTOM_COUNTS("Net.SpdySettingsCwnd",
2993                                       val, 1, 200, 100);
2994           if (total_bytes_received_ > 10 * 1024) {
2995             UMA_HISTOGRAM_CUSTOM_COUNTS("Net.SpdySettingsCwnd10K",
2996                                         val, 1, 200, 100);
2997             if (total_bytes_received_ > 25 * 1024) {
2998               UMA_HISTOGRAM_CUSTOM_COUNTS("Net.SpdySettingsCwnd25K",
2999                                           val, 1, 200, 100);
3000               if (total_bytes_received_ > 50 * 1024) {
3001                 UMA_HISTOGRAM_CUSTOM_COUNTS("Net.SpdySettingsCwnd50K",
3002                                             val, 1, 200, 100);
3003                 if (total_bytes_received_ > 100 * 1024) {
3004                   UMA_HISTOGRAM_CUSTOM_COUNTS("Net.SpdySettingsCwnd100K",
3005                                               val, 1, 200, 100);
3006                 }
3007               }
3008             }
3009           }
3010           break;
3011         case SETTINGS_ROUND_TRIP_TIME:
3012           UMA_HISTOGRAM_CUSTOM_COUNTS("Net.SpdySettingsRTT",
3013                                       val, 1, 1200, 100);
3014           break;
3015         case SETTINGS_DOWNLOAD_RETRANS_RATE:
3016           UMA_HISTOGRAM_CUSTOM_COUNTS("Net.SpdySettingsRetransRate",
3017                                       val, 1, 100, 50);
3018           break;
3019         default:
3020           break;
3021       }
3022     }
3023   }
3024 }
3025
3026 void SpdySession::CompleteStreamRequest(
3027     const base::WeakPtr<SpdyStreamRequest>& pending_request) {
3028   // Abort if the request has already been cancelled.
3029   if (!pending_request)
3030     return;
3031
3032   base::WeakPtr<SpdyStream> stream;
3033   int rv = TryCreateStream(pending_request, &stream);
3034
3035   if (rv == OK) {
3036     DCHECK(stream);
3037     pending_request->OnRequestCompleteSuccess(stream);
3038     return;
3039   }
3040   DCHECK(!stream);
3041
3042   if (rv != ERR_IO_PENDING) {
3043     pending_request->OnRequestCompleteFailure(rv);
3044   }
3045 }
3046
3047 SSLClientSocket* SpdySession::GetSSLClientSocket() const {
3048   if (!is_secure_)
3049     return NULL;
3050   SSLClientSocket* ssl_socket =
3051       reinterpret_cast<SSLClientSocket*>(connection_->socket());
3052   DCHECK(ssl_socket);
3053   return ssl_socket;
3054 }
3055
3056 void SpdySession::OnWriteBufferConsumed(
3057     size_t frame_payload_size,
3058     size_t consume_size,
3059     SpdyBuffer::ConsumeSource consume_source) {
3060   // We can be called with |in_io_loop_| set if a write SpdyBuffer is
3061   // deleted (e.g., a stream is closed due to incoming data).
3062
3063   DCHECK_EQ(flow_control_state_, FLOW_CONTROL_STREAM_AND_SESSION);
3064
3065   if (consume_source == SpdyBuffer::DISCARD) {
3066     // If we're discarding a frame or part of it, increase the send
3067     // window by the number of discarded bytes. (Although if we're
3068     // discarding part of a frame, it's probably because of a write
3069     // error and we'll be tearing down the session soon.)
3070     size_t remaining_payload_bytes = std::min(consume_size, frame_payload_size);
3071     DCHECK_GT(remaining_payload_bytes, 0u);
3072     IncreaseSendWindowSize(static_cast<int32>(remaining_payload_bytes));
3073   }
3074   // For consumed bytes, the send window is increased when we receive
3075   // a WINDOW_UPDATE frame.
3076 }
3077
3078 void SpdySession::IncreaseSendWindowSize(int32 delta_window_size) {
3079   // We can be called with |in_io_loop_| set if a SpdyBuffer is
3080   // deleted (e.g., a stream is closed due to incoming data).
3081
3082   DCHECK_EQ(flow_control_state_, FLOW_CONTROL_STREAM_AND_SESSION);
3083   DCHECK_GE(delta_window_size, 1);
3084
3085   // Check for overflow.
3086   int32 max_delta_window_size = kint32max - session_send_window_size_;
3087   if (delta_window_size > max_delta_window_size) {
3088     RecordProtocolErrorHistogram(PROTOCOL_ERROR_INVALID_WINDOW_UPDATE_SIZE);
3089     DoDrainSession(
3090         ERR_SPDY_PROTOCOL_ERROR,
3091         "Received WINDOW_UPDATE [delta: " +
3092             base::IntToString(delta_window_size) +
3093             "] for session overflows session_send_window_size_ [current: " +
3094             base::IntToString(session_send_window_size_) + "]");
3095     return;
3096   }
3097
3098   session_send_window_size_ += delta_window_size;
3099
3100   net_log_.AddEvent(
3101       NetLog::TYPE_SPDY_SESSION_UPDATE_SEND_WINDOW,
3102       base::Bind(&NetLogSpdySessionWindowUpdateCallback,
3103                  delta_window_size, session_send_window_size_));
3104
3105   DCHECK(!IsSendStalled());
3106   ResumeSendStalledStreams();
3107 }
3108
3109 void SpdySession::DecreaseSendWindowSize(int32 delta_window_size) {
3110   DCHECK_EQ(flow_control_state_, FLOW_CONTROL_STREAM_AND_SESSION);
3111
3112   // We only call this method when sending a frame. Therefore,
3113   // |delta_window_size| should be within the valid frame size range.
3114   DCHECK_GE(delta_window_size, 1);
3115   DCHECK_LE(delta_window_size, kMaxSpdyFrameChunkSize);
3116
3117   // |send_window_size_| should have been at least |delta_window_size| for
3118   // this call to happen.
3119   DCHECK_GE(session_send_window_size_, delta_window_size);
3120
3121   session_send_window_size_ -= delta_window_size;
3122
3123   net_log_.AddEvent(
3124       NetLog::TYPE_SPDY_SESSION_UPDATE_SEND_WINDOW,
3125       base::Bind(&NetLogSpdySessionWindowUpdateCallback,
3126                  -delta_window_size, session_send_window_size_));
3127 }
3128
3129 void SpdySession::OnReadBufferConsumed(
3130     size_t consume_size,
3131     SpdyBuffer::ConsumeSource consume_source) {
3132   // We can be called with |in_io_loop_| set if a read SpdyBuffer is
3133   // deleted (e.g., discarded by a SpdyReadQueue).
3134
3135   DCHECK_EQ(flow_control_state_, FLOW_CONTROL_STREAM_AND_SESSION);
3136   DCHECK_GE(consume_size, 1u);
3137   DCHECK_LE(consume_size, static_cast<size_t>(kint32max));
3138
3139   IncreaseRecvWindowSize(static_cast<int32>(consume_size));
3140 }
3141
3142 void SpdySession::IncreaseRecvWindowSize(int32 delta_window_size) {
3143   DCHECK_EQ(flow_control_state_, FLOW_CONTROL_STREAM_AND_SESSION);
3144   DCHECK_GE(session_unacked_recv_window_bytes_, 0);
3145   DCHECK_GE(session_recv_window_size_, session_unacked_recv_window_bytes_);
3146   DCHECK_GE(delta_window_size, 1);
3147   // Check for overflow.
3148   DCHECK_LE(delta_window_size, kint32max - session_recv_window_size_);
3149
3150   session_recv_window_size_ += delta_window_size;
3151   net_log_.AddEvent(
3152       NetLog::TYPE_SPDY_STREAM_UPDATE_RECV_WINDOW,
3153       base::Bind(&NetLogSpdySessionWindowUpdateCallback,
3154                  delta_window_size, session_recv_window_size_));
3155
3156   session_unacked_recv_window_bytes_ += delta_window_size;
3157   if (session_unacked_recv_window_bytes_ > kSpdySessionInitialWindowSize / 2) {
3158     SendWindowUpdateFrame(kSessionFlowControlStreamId,
3159                           session_unacked_recv_window_bytes_,
3160                           HIGHEST);
3161     session_unacked_recv_window_bytes_ = 0;
3162   }
3163 }
3164
3165 void SpdySession::DecreaseRecvWindowSize(int32 delta_window_size) {
3166   CHECK(in_io_loop_);
3167   DCHECK_EQ(flow_control_state_, FLOW_CONTROL_STREAM_AND_SESSION);
3168   DCHECK_GE(delta_window_size, 1);
3169
3170   // Since we never decrease the initial receive window size,
3171   // |delta_window_size| should never cause |recv_window_size_| to go
3172   // negative. If we do, the receive window isn't being respected.
3173   if (delta_window_size > session_recv_window_size_) {
3174     RecordProtocolErrorHistogram(PROTOCOL_ERROR_RECEIVE_WINDOW_VIOLATION);
3175     DoDrainSession(
3176         ERR_SPDY_FLOW_CONTROL_ERROR,
3177         "delta_window_size is " + base::IntToString(delta_window_size) +
3178             " in DecreaseRecvWindowSize, which is larger than the receive " +
3179             "window size of " + base::IntToString(session_recv_window_size_));
3180     return;
3181   }
3182
3183   session_recv_window_size_ -= delta_window_size;
3184   net_log_.AddEvent(
3185       NetLog::TYPE_SPDY_SESSION_UPDATE_RECV_WINDOW,
3186       base::Bind(&NetLogSpdySessionWindowUpdateCallback,
3187                  -delta_window_size, session_recv_window_size_));
3188 }
3189
3190 void SpdySession::QueueSendStalledStream(const SpdyStream& stream) {
3191   DCHECK(stream.send_stalled_by_flow_control());
3192   RequestPriority priority = stream.priority();
3193   CHECK_GE(priority, MINIMUM_PRIORITY);
3194   CHECK_LE(priority, MAXIMUM_PRIORITY);
3195   stream_send_unstall_queue_[priority].push_back(stream.stream_id());
3196 }
3197
3198 void SpdySession::ResumeSendStalledStreams() {
3199   DCHECK_EQ(flow_control_state_, FLOW_CONTROL_STREAM_AND_SESSION);
3200
3201   // We don't have to worry about new streams being queued, since
3202   // doing so would cause IsSendStalled() to return true. But we do
3203   // have to worry about streams being closed, as well as ourselves
3204   // being closed.
3205
3206   while (!IsSendStalled()) {
3207     size_t old_size = 0;
3208 #if DCHECK_IS_ON
3209     old_size = GetTotalSize(stream_send_unstall_queue_);
3210 #endif
3211
3212     SpdyStreamId stream_id = PopStreamToPossiblyResume();
3213     if (stream_id == 0)
3214       break;
3215     ActiveStreamMap::const_iterator it = active_streams_.find(stream_id);
3216     // The stream may actually still be send-stalled after this (due
3217     // to its own send window) but that's okay -- it'll then be
3218     // resumed once its send window increases.
3219     if (it != active_streams_.end())
3220       it->second.stream->PossiblyResumeIfSendStalled();
3221
3222     // The size should decrease unless we got send-stalled again.
3223     if (!IsSendStalled())
3224       DCHECK_LT(GetTotalSize(stream_send_unstall_queue_), old_size);
3225   }
3226 }
3227
3228 SpdyStreamId SpdySession::PopStreamToPossiblyResume() {
3229   for (int i = MAXIMUM_PRIORITY; i >= MINIMUM_PRIORITY; --i) {
3230     std::deque<SpdyStreamId>* queue = &stream_send_unstall_queue_[i];
3231     if (!queue->empty()) {
3232       SpdyStreamId stream_id = queue->front();
3233       queue->pop_front();
3234       return stream_id;
3235     }
3236   }
3237   return 0;
3238 }
3239
3240 }  // namespace net