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