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