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