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