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