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