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