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