net/spdy/spdy_session.cc

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