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