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