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 <map>
   8
   9 #include "base/basictypes.h"
  10 #include "base/bind.h"
  11 #include "base/compiler_specific.h"
  12 #include "base/logging.h"
  13 #include "base/message_loop.h"
  14 #include "base/metrics/field_trial.h"
  15 #include "base/metrics/histogram.h"
  16 #include "base/metrics/stats_counters.h"
  17 #include "base/stl_util.h"
  18 #include "base/string_number_conversions.h"
  19 #include "base/string_util.h"
  20 #include "base/stringprintf.h"
  21 #include "base/time.h"
  22 #include "base/utf_string_conversions.h"
  23 #include "base/values.h"
  24 #include "crypto/ec_private_key.h"
  25 #include "crypto/ec_signature_creator.h"
  26 #include "net/base/asn1_util.h"
  27 #include "net/base/connection_type_histograms.h"
  28 #include "net/base/net_log.h"
  29 #include "net/base/net_util.h"
  30 #include "net/base/server_bound_cert_service.h"
  31 #include "net/http/http_network_session.h"
  32 #include "net/http/http_server_properties.h"
  33 #include "net/spdy/spdy_credential_builder.h"
  34 #include "net/spdy/spdy_frame_builder.h"
  35 #include "net/spdy/spdy_http_utils.h"
  36 #include "net/spdy/spdy_protocol.h"
  37 #include "net/spdy/spdy_session_pool.h"
  38 #include "net/spdy/spdy_stream.h"
  39
  40 namespace net {
  41
  42 namespace {
  43
  44 const int kReadBufferSize = 8 * 1024;
  45 const int kDefaultConnectionAtRiskOfLossSeconds = 10;
  46 const int kHungIntervalSeconds = 10;
  47
  48 // Minimum seconds that unclaimed pushed streams will be kept in memory.
  49 const int kMinPushedStreamLifetimeSeconds = 300;
  50
  51 Value* NetLogSpdySynCallback(const SpdyHeaderBlock* headers,
  52                              bool fin,
  53                              bool unidirectional,
  54                              SpdyStreamId stream_id,
  55                              SpdyStreamId associated_stream,
  56                              NetLog::LogLevel /* log_level */) {
  57   DictionaryValue* dict = new DictionaryValue();
  58   ListValue* headers_list = new ListValue();
  59   for (SpdyHeaderBlock::const_iterator it = headers->begin();
  60        it != headers->end(); ++it) {
  61     headers_list->Append(new StringValue(base::StringPrintf(
  62         "%s: %s", it->first.c_str(), it->second.c_str())));
  63   }
  64   dict->SetBoolean("fin", fin);
  65   dict->SetBoolean("unidirectional", unidirectional);
  66   dict->Set("headers", headers_list);
  67   dict->SetInteger("stream_id", stream_id);
  68   if (associated_stream)
  69     dict->SetInteger("associated_stream", associated_stream);
  70   return dict;
  71 }
  72
  73 Value* NetLogSpdyCredentialCallback(size_t slot,
  74                                     const std::string* origin,
  75                                     NetLog::LogLevel /* log_level */) {
  76   DictionaryValue* dict = new DictionaryValue();
  77   dict->SetInteger("slot", slot);
  78   dict->SetString("origin", *origin);
  79   return dict;
  80 }
  81
  82 Value* NetLogSpdySessionCloseCallback(int net_error,
  83                                       const std::string* description,
  84                                       NetLog::LogLevel /* log_level */) {
  85   DictionaryValue* dict = new DictionaryValue();
  86   dict->SetInteger("net_error", net_error);
  87   dict->SetString("description", *description);
  88   return dict;
  89 }
  90
  91 Value* NetLogSpdySessionCallback(const HostPortProxyPair* host_pair,
  92                                  NetLog::LogLevel /* log_level */) {
  93   DictionaryValue* dict = new DictionaryValue();
  94   dict->SetString("host", host_pair->first.ToString());
  95   dict->SetString("proxy", host_pair->second.ToPacString());
  96   return dict;
  97 }
  98
  99 Value* NetLogSpdySettingCallback(SpdySettingsIds id,
 100                                  SpdySettingsFlags flags,
 101                                  uint32 value,
 102                                  NetLog::LogLevel /* log_level */) {
 103   DictionaryValue* dict = new DictionaryValue();
 104   dict->SetInteger("id", id);
 105   dict->SetInteger("flags", flags);
 106   dict->SetInteger("value", value);
 107   return dict;
 108 }
 109
 110 Value* NetLogSpdySettingsCallback(const SettingsMap* settings,
 111                                   NetLog::LogLevel /* log_level */) {
 112   DictionaryValue* dict = new DictionaryValue();
 113   ListValue* settings_list = new ListValue();
 114   for (SettingsMap::const_iterator it = settings->begin();
 115        it != settings->end(); ++it) {
 116     const SpdySettingsIds id = it->first;
 117     const SpdySettingsFlags flags = it->second.first;
 118     const uint32 value = it->second.second;
 119     settings_list->Append(new StringValue(
 120         base::StringPrintf("[id:%u flags:%u value:%u]", id, flags, value)));
 121   }
 122   dict->Set("settings", settings_list);
 123   return dict;
 124 }
 125
 126 Value* NetLogSpdyWindowUpdateCallback(SpdyStreamId stream_id,
 127                                       int32 delta,
 128                                       NetLog::LogLevel /* log_level */) {
 129   DictionaryValue* dict = new DictionaryValue();
 130   dict->SetInteger("stream_id", static_cast<int>(stream_id));
 131   dict->SetInteger("delta", delta);
 132   return dict;
 133 }
 134
 135 Value* NetLogSpdyDataCallback(SpdyStreamId stream_id,
 136                               int size,
 137                               SpdyDataFlags flags,
 138                               NetLog::LogLevel /* log_level */) {
 139   DictionaryValue* dict = new DictionaryValue();
 140   dict->SetInteger("stream_id", static_cast<int>(stream_id));
 141   dict->SetInteger("size", size);
 142   dict->SetInteger("flags", static_cast<int>(flags));
 143   return dict;
 144 }
 145
 146 Value* NetLogSpdyRstCallback(SpdyStreamId stream_id,
 147                              int status,
 148                              const std::string* description,
 149                              NetLog::LogLevel /* log_level */) {
 150   DictionaryValue* dict = new DictionaryValue();
 151   dict->SetInteger("stream_id", static_cast<int>(stream_id));
 152   dict->SetInteger("status", status);
 153   dict->SetString("description", *description);
 154   return dict;
 155 }
 156
 157 Value* NetLogSpdyPingCallback(uint32 unique_id,
 158                               const char* type,
 159                               NetLog::LogLevel /* log_level */) {
 160   DictionaryValue* dict = new DictionaryValue();
 161   dict->SetInteger("unique_id", unique_id);
 162   dict->SetString("type", type);
 163   return dict;
 164 }
 165
 166 Value* NetLogSpdyGoAwayCallback(SpdyStreamId last_stream_id,
 167                                 int active_streams,
 168                                 int unclaimed_streams,
 169                                 SpdyGoAwayStatus status,
 170                                 NetLog::LogLevel /* log_level */) {
 171   DictionaryValue* dict = new DictionaryValue();
 172   dict->SetInteger("last_accepted_stream_id",
 173                    static_cast<int>(last_stream_id));
 174   dict->SetInteger("active_streams", active_streams);
 175   dict->SetInteger("unclaimed_streams", unclaimed_streams);
 176   dict->SetInteger("status", static_cast<int>(status));
 177   return dict;
 178 }
 179
 180 // Maximum number of concurrent streams we will create, unless the server
 181 // sends a SETTINGS frame with a different value.
 182 const size_t kInitialMaxConcurrentStreams = 100;
 183 // The maximum number of concurrent streams we will ever create.  Even if
 184 // the server permits more, we will never exceed this limit.
 185 const size_t kMaxConcurrentStreamLimit = 256;
 186 const size_t kDefaultInitialRecvWindowSize = 10 * 1024 * 1024;  // 10MB
 187
 188 }  // namespace
 189
 190 // static
 191 void SpdySession::SpdyIOBufferProducer::ActivateStream(
 192     SpdySession* spdy_session,
 193     SpdyStream* spdy_stream) {
 194   spdy_session->ActivateStream(spdy_stream);
 195 }
 196
 197 // static
 198 SpdyIOBuffer* SpdySession::SpdyIOBufferProducer::CreateIOBuffer(
 199     SpdyFrame* frame,
 200     RequestPriority priority,
 201     SpdyStream* stream) {
 202   size_t size = frame->length() + SpdyFrame::kHeaderSize;
 203   DCHECK_GT(size, 0u);
 204
 205   // TODO(mbelshe): We have too much copying of data here.
 206   IOBufferWithSize* buffer = new IOBufferWithSize(size);
 207   memcpy(buffer->data(), frame->data(), size);
 208
 209   return new SpdyIOBuffer(buffer, size, priority, stream);
 210 }
 211
 212 SpdySession::SpdySession(const HostPortProxyPair& host_port_proxy_pair,
 213                          SpdySessionPool* spdy_session_pool,
 214                          HttpServerProperties* http_server_properties,
 215                          bool verify_domain_authentication,
 216                          bool enable_sending_initial_settings,
 217                          bool enable_credential_frames,
 218                          bool enable_compression,
 219                          bool enable_ping_based_connection_checking,
 220                          NextProto default_protocol,
 221                          size_t initial_recv_window_size,
 222                          size_t initial_max_concurrent_streams,
 223                          size_t max_concurrent_streams_limit,
 224                          TimeFunc time_func,
 225                          const HostPortPair& trusted_spdy_proxy,
 226                          NetLog* net_log)
 227     : ALLOW_THIS_IN_INITIALIZER_LIST(weak_factory_(this)),
 228       host_port_proxy_pair_(host_port_proxy_pair),
 229       spdy_session_pool_(spdy_session_pool),
 230       http_server_properties_(http_server_properties),
 231       connection_(new ClientSocketHandle),
 232       read_buffer_(new IOBuffer(kReadBufferSize)),
 233       read_pending_(false),
 234       stream_hi_water_mark_(1),  // Always start at 1 for the first stream id.
 235       write_pending_(false),
 236       delayed_write_pending_(false),
 237       is_secure_(false),
 238       certificate_error_code_(OK),
 239       error_(OK),
 240       state_(IDLE),
 241       max_concurrent_streams_(initial_max_concurrent_streams == 0 ?
 242                               kInitialMaxConcurrentStreams :
 243                               initial_max_concurrent_streams),
 244       max_concurrent_streams_limit_(max_concurrent_streams_limit == 0 ?
 245                                     kMaxConcurrentStreamLimit :
 246                                     max_concurrent_streams_limit),
 247       streams_initiated_count_(0),
 248       streams_pushed_count_(0),
 249       streams_pushed_and_claimed_count_(0),
 250       streams_abandoned_count_(0),
 251       bytes_received_(0),
 252       sent_settings_(false),
 253       received_settings_(false),
 254       stalled_streams_(0),
 255       pings_in_flight_(0),
 256       next_ping_id_(1),
 257       last_activity_time_(base::TimeTicks::Now()),
 258       check_ping_status_pending_(false),
 259       flow_control_(false),
 260       initial_send_window_size_(kSpdyStreamInitialWindowSize),
 261       initial_recv_window_size_(initial_recv_window_size == 0 ?
 262                                 kDefaultInitialRecvWindowSize :
 263                                 initial_recv_window_size),
 264       net_log_(BoundNetLog::Make(net_log, NetLog::SOURCE_SPDY_SESSION)),
 265       verify_domain_authentication_(verify_domain_authentication),
 266       enable_sending_initial_settings_(enable_sending_initial_settings),
 267       enable_credential_frames_(enable_credential_frames),
 268       enable_compression_(enable_compression),
 269       enable_ping_based_connection_checking_(
 270           enable_ping_based_connection_checking),
 271       default_protocol_(default_protocol),
 272       credential_state_(SpdyCredentialState::kDefaultNumSlots),
 273       connection_at_risk_of_loss_time_(
 274           base::TimeDelta::FromSeconds(kDefaultConnectionAtRiskOfLossSeconds)),
 275       hung_interval_(
 276           base::TimeDelta::FromSeconds(kHungIntervalSeconds)),
 277       trusted_spdy_proxy_(trusted_spdy_proxy),
 278       time_func_(time_func) {
 279   DCHECK(HttpStreamFactory::spdy_enabled());
 280   net_log_.BeginEvent(
 281       NetLog::TYPE_SPDY_SESSION,
 282       base::Bind(&NetLogSpdySessionCallback, &host_port_proxy_pair_));
 283   next_unclaimed_push_stream_sweep_time_ = time_func_() +
 284       base::TimeDelta::FromSeconds(kMinPushedStreamLifetimeSeconds);
 285   // TODO(mbelshe): consider randomization of the stream_hi_water_mark.
 286 }
 287
 288 SpdySession::PendingCreateStream::PendingCreateStream(
 289     const GURL& url, RequestPriority priority,
 290     scoped_refptr<SpdyStream>* spdy_stream,
 291     const BoundNetLog& stream_net_log,
 292     const CompletionCallback& callback)
 293     : url(&url),
 294       priority(priority),
 295       spdy_stream(spdy_stream),
 296       stream_net_log(&stream_net_log),
 297       callback(callback) {
 298 }
 299
 300 SpdySession::PendingCreateStream::~PendingCreateStream() {}
 301
 302 SpdySession::CallbackResultPair::CallbackResultPair(
 303     const CompletionCallback& callback_in, int result_in)
 304     : callback(callback_in),
 305       result(result_in) {
 306 }
 307
 308 SpdySession::CallbackResultPair::~CallbackResultPair() {}
 309
 310 SpdySession::~SpdySession() {
 311   if (state_ != CLOSED) {
 312     state_ = CLOSED;
 313
 314     // Cleanup all the streams.
 315     CloseAllStreams(net::ERR_ABORTED);
 316   }
 317
 318   if (connection_->is_initialized()) {
 319     // With SPDY we can't recycle sockets.
 320     connection_->socket()->Disconnect();
 321   }
 322
 323   // Streams should all be gone now.
 324   DCHECK_EQ(0u, num_active_streams());
 325   DCHECK_EQ(0u, num_unclaimed_pushed_streams());
 326
 327   DCHECK(pending_callback_map_.empty());
 328
 329   RecordHistograms();
 330
 331   net_log_.EndEvent(NetLog::TYPE_SPDY_SESSION);
 332 }
 333
 334 net::Error SpdySession::InitializeWithSocket(
 335     ClientSocketHandle* connection,
 336     bool is_secure,
 337     int certificate_error_code) {
 338   base::StatsCounter spdy_sessions("spdy.sessions");
 339   spdy_sessions.Increment();
 340
 341   state_ = CONNECTED;
 342   connection_.reset(connection);
 343   is_secure_ = is_secure;
 344   certificate_error_code_ = certificate_error_code;
 345
 346   NextProto protocol = default_protocol_;
 347   NextProto protocol_negotiated = connection->socket()->GetNegotiatedProtocol();
 348   if (protocol_negotiated != kProtoUnknown) {
 349     protocol = protocol_negotiated;
 350   }
 351
 352   SSLClientSocket* ssl_socket = GetSSLClientSocket();
 353   if (ssl_socket && ssl_socket->WasChannelIDSent()) {
 354     // According to the SPDY spec, the credential associated with the TLS
 355     // connection is stored in slot[1].
 356     credential_state_.SetHasCredential(GURL("https://" +
 357                                             host_port_pair().ToString()));
 358   }
 359
 360   DCHECK(protocol >= kProtoSPDY2);
 361   DCHECK(protocol <= kProtoSPDY3);
 362   int version = (protocol == kProtoSPDY3) ? 3 : 2;
 363   flow_control_ = (protocol >= kProtoSPDY3);
 364
 365   buffered_spdy_framer_.reset(new BufferedSpdyFramer(version,
 366                                                      enable_compression_));
 367   buffered_spdy_framer_->set_visitor(this);
 368   SendInitialSettings();
 369   UMA_HISTOGRAM_ENUMERATION("Net.SpdyVersion", protocol, kProtoMaximumVersion);
 370
 371   // Write out any data that we might have to send, such as the settings frame.
 372   WriteSocketLater();
 373   net::Error error = ReadSocket();
 374   if (error == ERR_IO_PENDING)
 375     return OK;
 376   return error;
 377 }
 378
 379 bool SpdySession::VerifyDomainAuthentication(const std::string& domain) {
 380   if (!verify_domain_authentication_)
 381     return true;
 382
 383   if (state_ != CONNECTED)
 384     return false;
 385
 386   SSLInfo ssl_info;
 387   bool was_npn_negotiated;
 388   NextProto protocol_negotiated = kProtoUnknown;
 389   if (!GetSSLInfo(&ssl_info, &was_npn_negotiated, &protocol_negotiated))
 390     return true;   // This is not a secure session, so all domains are okay.
 391
 392   return !ssl_info.client_cert_sent &&
 393       (enable_credential_frames_ || !ssl_info.channel_id_sent ||
 394        ServerBoundCertService::GetDomainForHost(domain) ==
 395        ServerBoundCertService::GetDomainForHost(
 396            host_port_proxy_pair_.first.host())) &&
 397       ssl_info.cert->VerifyNameMatch(domain);
 398 }
 399
 400 void SpdySession::SetStreamHasWriteAvailable(SpdyStream* stream,
 401                                              SpdyIOBufferProducer* producer) {
 402   write_queue_.push(producer);
 403   stream_producers_[producer] = stream;
 404   WriteSocketLater();
 405 }
 406
 407 int SpdySession::GetPushStream(
 408     const GURL& url,
 409     scoped_refptr<SpdyStream>* stream,
 410     const BoundNetLog& stream_net_log) {
 411   CHECK_NE(state_, CLOSED);
 412
 413   *stream = NULL;
 414
 415   // Don't allow access to secure push streams over an unauthenticated, but
 416   // encrypted SSL socket.
 417   if (is_secure_ && certificate_error_code_ != OK &&
 418       (url.SchemeIs("https") || url.SchemeIs("wss"))) {
 419     RecordProtocolErrorHistogram(
 420         PROTOCOL_ERROR_REQUEST_FOR_SECURE_CONTENT_OVER_INSECURE_SESSION);
 421     CloseSessionOnError(
 422         static_cast<net::Error>(certificate_error_code_),
 423         true,
 424         "Tried to get SPDY stream for secure content over an unauthenticated "
 425         "session.");
 426     return ERR_SPDY_PROTOCOL_ERROR;
 427   }
 428
 429   *stream = GetActivePushStream(url.spec());
 430   if (stream->get()) {
 431     DCHECK(streams_pushed_and_claimed_count_ < streams_pushed_count_);
 432     streams_pushed_and_claimed_count_++;
 433     return OK;
 434   }
 435   return 0;
 436 }
 437
 438 int SpdySession::CreateStream(
 439     const GURL& url,
 440     RequestPriority priority,
 441     scoped_refptr<SpdyStream>* spdy_stream,
 442     const BoundNetLog& stream_net_log,
 443     const CompletionCallback& callback) {
 444   if (!max_concurrent_streams_ ||
 445       (active_streams_.size() + created_streams_.size() <
 446        max_concurrent_streams_)) {
 447     return CreateStreamImpl(url, priority, spdy_stream, stream_net_log);
 448   }
 449
 450   stalled_streams_++;
 451   net_log().AddEvent(NetLog::TYPE_SPDY_SESSION_STALLED_MAX_STREAMS);
 452   create_stream_queues_[priority].push(
 453       PendingCreateStream(url, priority, spdy_stream,
 454                           stream_net_log, callback));
 455   return ERR_IO_PENDING;
 456 }
 457
 458 void SpdySession::ProcessPendingCreateStreams() {
 459   while (!max_concurrent_streams_ ||
 460          active_streams_.size() < max_concurrent_streams_) {
 461     bool no_pending_create_streams = true;
 462     for (int i = NUM_PRIORITIES - 1; i >= MINIMUM_PRIORITY; --i) {
 463       if (!create_stream_queues_[i].empty()) {
 464         PendingCreateStream pending_create = create_stream_queues_[i].front();
 465         create_stream_queues_[i].pop();
 466         no_pending_create_streams = false;
 467         int error = CreateStreamImpl(*pending_create.url,
 468                                      pending_create.priority,
 469                                      pending_create.spdy_stream,
 470                                      *pending_create.stream_net_log);
 471         scoped_refptr<SpdyStream>* stream = pending_create.spdy_stream;
 472         DCHECK(!ContainsKey(pending_callback_map_, stream));
 473         pending_callback_map_.insert(std::make_pair(stream,
 474             CallbackResultPair(pending_create.callback, error)));
 475         MessageLoop::current()->PostTask(
 476             FROM_HERE,
 477             base::Bind(&SpdySession::InvokeUserStreamCreationCallback,
 478                        weak_factory_.GetWeakPtr(), stream));
 479         break;
 480       }
 481     }
 482     if (no_pending_create_streams)
 483       return;  // there were no streams in any queue
 484   }
 485 }
 486
 487 void SpdySession::CancelPendingCreateStreams(
 488     const scoped_refptr<SpdyStream>* spdy_stream) {
 489   PendingCallbackMap::iterator it = pending_callback_map_.find(spdy_stream);
 490   if (it != pending_callback_map_.end()) {
 491     pending_callback_map_.erase(it);
 492     return;
 493   }
 494
 495   for (int i = 0; i < NUM_PRIORITIES; ++i) {
 496     PendingCreateStreamQueue tmp;
 497     // Make a copy removing this trans
 498     while (!create_stream_queues_[i].empty()) {
 499       PendingCreateStream pending_create = create_stream_queues_[i].front();
 500       create_stream_queues_[i].pop();
 501       if (pending_create.spdy_stream != spdy_stream)
 502         tmp.push(pending_create);
 503     }
 504     // Now copy it back
 505     while (!tmp.empty()) {
 506       create_stream_queues_[i].push(tmp.front());
 507       tmp.pop();
 508     }
 509   }
 510 }
 511
 512 int SpdySession::CreateStreamImpl(
 513     const GURL& url,
 514     RequestPriority priority,
 515     scoped_refptr<SpdyStream>* spdy_stream,
 516     const BoundNetLog& stream_net_log) {
 517   DCHECK_GE(priority, MINIMUM_PRIORITY);
 518   DCHECK_LT(priority, NUM_PRIORITIES);
 519
 520   // Make sure that we don't try to send https/wss over an unauthenticated, but
 521   // encrypted SSL socket.
 522   if (is_secure_ && certificate_error_code_ != OK &&
 523       (url.SchemeIs("https") || url.SchemeIs("wss"))) {
 524     RecordProtocolErrorHistogram(
 525         PROTOCOL_ERROR_REQUEST_FOR_SECURE_CONTENT_OVER_INSECURE_SESSION);
 526     CloseSessionOnError(
 527         static_cast<net::Error>(certificate_error_code_),
 528         true,
 529         "Tried to create SPDY stream for secure content over an "
 530         "unauthenticated session.");
 531     return ERR_SPDY_PROTOCOL_ERROR;
 532   }
 533
 534   const std::string& path = url.PathForRequest();
 535
 536   *spdy_stream = new SpdyStream(this,
 537                                 false,
 538                                 stream_net_log);
 539   const scoped_refptr<SpdyStream>& stream = *spdy_stream;
 540
 541   stream->set_priority(priority);
 542   stream->set_path(path);
 543   stream->set_send_window_size(initial_send_window_size_);
 544   stream->set_recv_window_size(initial_recv_window_size_);
 545   created_streams_.insert(stream);
 546
 547   UMA_HISTOGRAM_CUSTOM_COUNTS("Net.SpdyPriorityCount",
 548       static_cast<int>(priority), 0, 10, 11);
 549
 550   // TODO(mbelshe): Optimize memory allocations
 551
 552   return OK;
 553 }
 554
 555 bool SpdySession::NeedsCredentials() const {
 556   if (!is_secure_)
 557     return false;
 558   SSLClientSocket* ssl_socket = GetSSLClientSocket();
 559   if (ssl_socket->GetNegotiatedProtocol() < kProtoSPDY3)
 560     return false;
 561   return ssl_socket->WasChannelIDSent();
 562 }
 563
 564 void SpdySession::AddPooledAlias(const HostPortProxyPair& alias) {
 565   pooled_aliases_.insert(alias);
 566 }
 567
 568 int SpdySession::GetProtocolVersion() const {
 569   DCHECK(buffered_spdy_framer_.get());
 570   return buffered_spdy_framer_->protocol_version();
 571 }
 572
 573 SpdySynStreamControlFrame* SpdySession::CreateSynStream(
 574     SpdyStreamId stream_id,
 575     RequestPriority priority,
 576     uint8 credential_slot,
 577     SpdyControlFlags flags,
 578     const SpdyHeaderBlock& headers) {
 579   CHECK(IsStreamActive(stream_id));
 580   const scoped_refptr<SpdyStream>& stream = active_streams_[stream_id];
 581   CHECK_EQ(stream->stream_id(), stream_id);
 582
 583   SendPrefacePingIfNoneInFlight();
 584
 585   DCHECK(buffered_spdy_framer_.get());
 586   scoped_ptr<SpdySynStreamControlFrame> syn_frame(
 587       buffered_spdy_framer_->CreateSynStream(
 588           stream_id, 0,
 589           ConvertRequestPriorityToSpdyPriority(priority, GetProtocolVersion()),
 590           credential_slot, flags, true, &headers));
 591
 592   base::StatsCounter spdy_requests("spdy.requests");
 593   spdy_requests.Increment();
 594   streams_initiated_count_++;
 595
 596   if (net_log().IsLoggingAllEvents()) {
 597     net_log().AddEvent(
 598         NetLog::TYPE_SPDY_SESSION_SYN_STREAM,
 599         base::Bind(&NetLogSpdySynCallback, &headers,
 600                    (flags & CONTROL_FLAG_FIN) != 0,
 601                    (flags & CONTROL_FLAG_UNIDIRECTIONAL) != 0,
 602                    stream_id, 0));
 603   }
 604
 605   return syn_frame.release();
 606 }
 607
 608 SpdyCredentialControlFrame* SpdySession::CreateCredentialFrame(
 609     const std::string& origin,
 610     SSLClientCertType type,
 611     const std::string& key,
 612     const std::string& cert,
 613     RequestPriority priority) {
 614   DCHECK(is_secure_);
 615   SSLClientSocket* ssl_socket = GetSSLClientSocket();
 616   DCHECK(ssl_socket);
 617   DCHECK(ssl_socket->WasChannelIDSent());
 618
 619   SpdyCredential credential;
 620   std::string tls_unique;
 621   ssl_socket->GetTLSUniqueChannelBinding(&tls_unique);
 622   size_t slot = credential_state_.SetHasCredential(GURL(origin));
 623   int rv = SpdyCredentialBuilder::Build(tls_unique, type, key, cert, slot,
 624                                         &credential);
 625   DCHECK_EQ(OK, rv);
 626   if (rv != OK)
 627     return NULL;
 628
 629   DCHECK(buffered_spdy_framer_.get());
 630   scoped_ptr<SpdyCredentialControlFrame> credential_frame(
 631       buffered_spdy_framer_->CreateCredentialFrame(credential));
 632
 633   if (net_log().IsLoggingAllEvents()) {
 634     net_log().AddEvent(
 635         NetLog::TYPE_SPDY_SESSION_SEND_CREDENTIAL,
 636         base::Bind(&NetLogSpdyCredentialCallback, credential.slot, &origin));
 637   }
 638   return credential_frame.release();
 639 }
 640
 641 SpdyHeadersControlFrame* SpdySession::CreateHeadersFrame(
 642     SpdyStreamId stream_id,
 643     const SpdyHeaderBlock& headers,
 644     SpdyControlFlags flags) {
 645   // Find our stream
 646   CHECK(IsStreamActive(stream_id));
 647   scoped_refptr<SpdyStream> stream = active_streams_[stream_id];
 648   CHECK_EQ(stream->stream_id(), stream_id);
 649
 650   // Create a HEADER frame.
 651   scoped_ptr<SpdyHeadersControlFrame> frame(
 652       buffered_spdy_framer_->CreateHeaders(stream_id, flags, true, &headers));
 653
 654   if (net_log().IsLoggingAllEvents()) {
 655     bool fin = flags & CONTROL_FLAG_FIN;
 656     net_log().AddEvent(
 657         NetLog::TYPE_SPDY_SESSION_SEND_HEADERS,
 658         base::Bind(&NetLogSpdySynCallback,
 659                    &headers, fin, /*unidirectional=*/false,
 660                    stream_id, 0));
 661   }
 662   return frame.release();
 663 }
 664
 665 SpdyDataFrame* SpdySession::CreateDataFrame(SpdyStreamId stream_id,
 666                                             net::IOBuffer* data, int len,
 667                                             SpdyDataFlags flags) {
 668   // Find our stream
 669   CHECK(IsStreamActive(stream_id));
 670   scoped_refptr<SpdyStream> stream = active_streams_[stream_id];
 671   CHECK_EQ(stream->stream_id(), stream_id);
 672
 673   if (len > kMaxSpdyFrameChunkSize) {
 674     len = kMaxSpdyFrameChunkSize;
 675     flags = static_cast<SpdyDataFlags>(flags & ~DATA_FLAG_FIN);
 676   }
 677
 678   // Obey send window size of the stream if flow control is enabled.
 679   if (flow_control_) {
 680     if (stream->send_window_size() <= 0) {
 681       // Because we queue frames onto the session, it is possible that
 682       // a stream was not flow controlled at the time it attempted the
 683       // write, but when we go to fulfill the write, it is now flow
 684       // controlled.  This is why we need the session to mark the stream
 685       // as stalled - because only the session knows for sure when the
 686       // stall occurs.
 687       stream->set_stalled_by_flow_control(true);
 688       net_log().AddEvent(
 689           NetLog::TYPE_SPDY_SESSION_STALLED_ON_SEND_WINDOW,
 690           NetLog::IntegerCallback("stream_id", stream_id));
 691       return NULL;
 692     }
 693     int new_len = std::min(len, stream->send_window_size());
 694     if (new_len < len) {
 695       len = new_len;
 696       flags = static_cast<SpdyDataFlags>(flags & ~DATA_FLAG_FIN);
 697     }
 698     stream->DecreaseSendWindowSize(len);
 699   }
 700
 701   if (net_log().IsLoggingAllEvents()) {
 702     net_log().AddEvent(
 703         NetLog::TYPE_SPDY_SESSION_SEND_DATA,
 704         base::Bind(&NetLogSpdyDataCallback, stream_id, len, flags));
 705   }
 706
 707   // Send PrefacePing for DATA_FRAMEs with nonzero payload size.
 708   if (len > 0)
 709     SendPrefacePingIfNoneInFlight();
 710
 711   // TODO(mbelshe): reduce memory copies here.
 712   DCHECK(buffered_spdy_framer_.get());
 713   scoped_ptr<SpdyDataFrame> frame(
 714       buffered_spdy_framer_->CreateDataFrame(
 715           stream_id, data->data(), len, flags));
 716
 717   return frame.release();
 718 }
 719
 720 void SpdySession::CloseStream(SpdyStreamId stream_id, int status) {
 721   DCHECK_NE(0u, stream_id);
 722   // TODO(mbelshe): We should send a RST_STREAM control frame here
 723   //                so that the server can cancel a large send.
 724
 725   DeleteStream(stream_id, status);
 726 }
 727
 728 void SpdySession::CloseCreatedStream(SpdyStream* stream, int status) {
 729   DCHECK_EQ(0u, stream->stream_id());
 730   created_streams_.erase(scoped_refptr<SpdyStream>(stream));
 731 }
 732
 733 void SpdySession::ResetStream(SpdyStreamId stream_id,
 734                               SpdyStatusCodes status,
 735                               const std::string& description) {
 736   net_log().AddEvent(
 737       NetLog::TYPE_SPDY_SESSION_SEND_RST_STREAM,
 738       base::Bind(&NetLogSpdyRstCallback, stream_id, status, &description));
 739
 740   DCHECK(buffered_spdy_framer_.get());
 741   scoped_ptr<SpdyRstStreamControlFrame> rst_frame(
 742       buffered_spdy_framer_->CreateRstStream(stream_id, status));
 743
 744   // Default to lowest priority unless we know otherwise.
 745   RequestPriority priority = net::IDLE;
 746   if(IsStreamActive(stream_id)) {
 747     scoped_refptr<SpdyStream> stream = active_streams_[stream_id];
 748     priority = stream->priority();
 749   }
 750   QueueFrame(rst_frame.release(), priority);
 751   RecordProtocolErrorHistogram(
 752       static_cast<SpdyProtocolErrorDetails>(status + STATUS_CODE_INVALID));
 753   DeleteStream(stream_id, ERR_SPDY_PROTOCOL_ERROR);
 754 }
 755
 756 bool SpdySession::IsStreamActive(SpdyStreamId stream_id) const {
 757   return ContainsKey(active_streams_, stream_id);
 758 }
 759
 760 LoadState SpdySession::GetLoadState() const {
 761   // NOTE: The application only queries the LoadState via the
 762   //       SpdyNetworkTransaction, and details are only needed when
 763   //       we're in the process of connecting.
 764
 765   // If we're connecting, defer to the connection to give us the actual
 766   // LoadState.
 767   if (state_ == CONNECTING)
 768     return connection_->GetLoadState();
 769
 770   // Just report that we're idle since the session could be doing
 771   // many things concurrently.
 772   return LOAD_STATE_IDLE;
 773 }
 774
 775 void SpdySession::OnReadComplete(int bytes_read) {
 776   // Parse a frame.  For now this code requires that the frame fit into our
 777   // buffer (32KB).
 778   // TODO(mbelshe): support arbitrarily large frames!
 779
 780   read_pending_ = false;
 781
 782   if (bytes_read <= 0) {
 783     // Session is tearing down.
 784     net::Error error = static_cast<net::Error>(bytes_read);
 785     if (bytes_read == 0)
 786       error = ERR_CONNECTION_CLOSED;
 787     CloseSessionOnError(error, true, "bytes_read is <= 0.");
 788     return;
 789   }
 790
 791   bytes_received_ += bytes_read;
 792
 793   last_activity_time_ = base::TimeTicks::Now();
 794
 795   // The SpdyFramer will use callbacks onto |this| as it parses frames.
 796   // When errors occur, those callbacks can lead to teardown of all references
 797   // to |this|, so maintain a reference to self during this call for safe
 798   // cleanup.
 799   scoped_refptr<SpdySession> self(this);
 800
 801   DCHECK(buffered_spdy_framer_.get());
 802   char *data = read_buffer_->data();
 803   while (bytes_read &&
 804          buffered_spdy_framer_->error_code() ==
 805              SpdyFramer::SPDY_NO_ERROR) {
 806     uint32 bytes_processed =
 807         buffered_spdy_framer_->ProcessInput(data, bytes_read);
 808     bytes_read -= bytes_processed;
 809     data += bytes_processed;
 810     if (buffered_spdy_framer_->state() == SpdyFramer::SPDY_DONE)
 811       buffered_spdy_framer_->Reset();
 812   }
 813
 814   if (state_ != CLOSED)
 815     ReadSocket();
 816 }
 817
 818 void SpdySession::OnWriteComplete(int result) {
 819   DCHECK(write_pending_);
 820   DCHECK(in_flight_write_.size());
 821
 822   last_activity_time_ = base::TimeTicks::Now();
 823   write_pending_ = false;
 824
 825   scoped_refptr<SpdyStream> stream = in_flight_write_.stream();
 826
 827   if (result >= 0) {
 828     // It should not be possible to have written more bytes than our
 829     // in_flight_write_.
 830     DCHECK_LE(result, in_flight_write_.buffer()->BytesRemaining());
 831
 832     in_flight_write_.buffer()->DidConsume(result);
 833
 834     // We only notify the stream when we've fully written the pending frame.
 835     if (!in_flight_write_.buffer()->BytesRemaining()) {
 836       if (stream) {
 837         // Report the number of bytes written to the caller, but exclude the
 838         // frame size overhead.  NOTE: if this frame was compressed the
 839         // reported bytes written is the compressed size, not the original
 840         // size.
 841         if (result > 0) {
 842           result = in_flight_write_.buffer()->size();
 843           DCHECK_GE(result, static_cast<int>(SpdyFrame::kHeaderSize));
 844           result -= static_cast<int>(SpdyFrame::kHeaderSize);
 845         }
 846
 847         // It is possible that the stream was cancelled while we were writing
 848         // to the socket.
 849         if (!stream->cancelled())
 850           stream->OnWriteComplete(result);
 851       }
 852
 853       // Cleanup the write which just completed.
 854       in_flight_write_.release();
 855     }
 856
 857     // Write more data.  We're already in a continuation, so we can
 858     // go ahead and write it immediately (without going back to the
 859     // message loop).
 860     WriteSocketLater();
 861   } else {
 862     in_flight_write_.release();
 863
 864     // The stream is now errored.  Close it down.
 865     CloseSessionOnError(
 866         static_cast<net::Error>(result), true, "The stream has errored.");
 867   }
 868 }
 869
 870 net::Error SpdySession::ReadSocket() {
 871   if (read_pending_)
 872     return OK;
 873
 874   if (state_ == CLOSED) {
 875     NOTREACHED();
 876     return ERR_UNEXPECTED;
 877   }
 878
 879   CHECK(connection_.get());
 880   CHECK(connection_->socket());
 881   int bytes_read = connection_->socket()->Read(
 882       read_buffer_.get(),
 883       kReadBufferSize,
 884       base::Bind(&SpdySession::OnReadComplete, base::Unretained(this)));
 885   switch (bytes_read) {
 886     case 0:
 887       // Socket is closed!
 888       CloseSessionOnError(ERR_CONNECTION_CLOSED, true, "bytes_read is 0.");
 889       return ERR_CONNECTION_CLOSED;
 890     case net::ERR_IO_PENDING:
 891       // Waiting for data.  Nothing to do now.
 892       read_pending_ = true;
 893       return ERR_IO_PENDING;
 894     default:
 895       // Data was read, process it.
 896       // Schedule the work through the message loop to avoid recursive
 897       // callbacks.
 898       read_pending_ = true;
 899       MessageLoop::current()->PostTask(
 900           FROM_HERE,
 901           base::Bind(&SpdySession::OnReadComplete,
 902                      weak_factory_.GetWeakPtr(), bytes_read));
 903       break;
 904   }
 905   return OK;
 906 }
 907
 908 void SpdySession::WriteSocketLater() {
 909   if (delayed_write_pending_)
 910     return;
 911
 912   if (state_ < CONNECTED)
 913     return;
 914
 915   delayed_write_pending_ = true;
 916   MessageLoop::current()->PostTask(
 917       FROM_HERE,
 918       base::Bind(&SpdySession::WriteSocket, weak_factory_.GetWeakPtr()));
 919 }
 920
 921 void SpdySession::WriteSocket() {
 922   // This function should only be called via WriteSocketLater.
 923   DCHECK(delayed_write_pending_);
 924   delayed_write_pending_ = false;
 925
 926   // If the socket isn't connected yet, just wait; we'll get called
 927   // again when the socket connection completes.  If the socket is
 928   // closed, just return.
 929   if (state_ < CONNECTED || state_ == CLOSED)
 930     return;
 931
 932   if (write_pending_)   // Another write is in progress still.
 933     return;
 934
 935   // Loop sending frames until we've sent everything or until the write
 936   // returns error (or ERR_IO_PENDING).
 937   DCHECK(buffered_spdy_framer_.get());
 938   while (in_flight_write_.buffer() || !write_queue_.empty()) {
 939     if (!in_flight_write_.buffer()) {
 940       // Grab the next SpdyBuffer to send.
 941       scoped_ptr<SpdyIOBufferProducer> producer(write_queue_.top());
 942       write_queue_.pop();
 943       scoped_ptr<SpdyIOBuffer> buffer(producer->ProduceNextBuffer(this));
 944       stream_producers_.erase(producer.get());
 945       // It is possible that a stream had data to write, but a
 946       // WINDOW_UPDATE frame has been received which made that
 947       // stream no longer writable.
 948       // TODO(rch): consider handling that case by removing the
 949       // stream from the writable queue?
 950       if (buffer == NULL)
 951         continue;
 952
 953       in_flight_write_ = *buffer;
 954     } else {
 955       DCHECK(in_flight_write_.buffer()->BytesRemaining());
 956     }
 957
 958     write_pending_ = true;
 959     int rv = connection_->socket()->Write(
 960         in_flight_write_.buffer(),
 961         in_flight_write_.buffer()->BytesRemaining(),
 962         base::Bind(&SpdySession::OnWriteComplete, base::Unretained(this)));
 963     if (rv == net::ERR_IO_PENDING)
 964       break;
 965
 966     // We sent the frame successfully.
 967     OnWriteComplete(rv);
 968
 969     // TODO(mbelshe):  Test this error case.  Maybe we should mark the socket
 970     //                 as in an error state.
 971     if (rv < 0)
 972       break;
 973   }
 974 }
 975
 976 void SpdySession::CloseAllStreams(net::Error status) {
 977   base::StatsCounter abandoned_streams("spdy.abandoned_streams");
 978   base::StatsCounter abandoned_push_streams(
 979       "spdy.abandoned_push_streams");
 980
 981   if (!active_streams_.empty())
 982     abandoned_streams.Add(active_streams_.size());
 983   if (!unclaimed_pushed_streams_.empty()) {
 984     streams_abandoned_count_ += unclaimed_pushed_streams_.size();
 985     abandoned_push_streams.Add(unclaimed_pushed_streams_.size());
 986     unclaimed_pushed_streams_.clear();
 987   }
 988
 989   for (int i = 0; i < NUM_PRIORITIES; ++i) {
 990     while (!create_stream_queues_[i].empty()) {
 991       PendingCreateStream pending_create = create_stream_queues_[i].front();
 992       create_stream_queues_[i].pop();
 993       pending_create.callback.Run(ERR_ABORTED);
 994     }
 995   }
 996
 997   while (!active_streams_.empty()) {
 998     ActiveStreamMap::iterator it = active_streams_.begin();
 999     const scoped_refptr<SpdyStream>& stream = it->second;
1000     LogAbandonedStream(stream, status);
1001     DeleteStream(stream->stream_id(), status);
1002   }
1003
1004   while (!created_streams_.empty()) {
1005     CreatedStreamSet::iterator it = created_streams_.begin();
1006     const scoped_refptr<SpdyStream> stream = *it;
1007     created_streams_.erase(it);
1008     LogAbandonedStream(stream, status);
1009     stream->OnClose(status);
1010   }
1011
1012   // We also need to drain the queue.
1013   while (!write_queue_.empty()) {
1014     scoped_ptr<SpdyIOBufferProducer> producer(write_queue_.top());
1015     write_queue_.pop();
1016     stream_producers_.erase(producer.get());
1017   }
1018 }
1019
1020 void SpdySession::LogAbandonedStream(const scoped_refptr<SpdyStream>& stream,
1021                                      net::Error status) {
1022   DCHECK(stream);
1023   std::string description = base::StringPrintf(
1024       "ABANDONED (stream_id=%d): ", stream->stream_id()) + stream->path();
1025   stream->LogStreamError(status, description);
1026 }
1027
1028 int SpdySession::GetNewStreamId() {
1029   int id = stream_hi_water_mark_;
1030   stream_hi_water_mark_ += 2;
1031   if (stream_hi_water_mark_ > 0x7fff)
1032     stream_hi_water_mark_ = 1;
1033   return id;
1034 }
1035
1036 void SpdySession::CloseSessionOnError(net::Error err,
1037                                       bool remove_from_pool,
1038                                       const std::string& description) {
1039   // Closing all streams can have a side-effect of dropping the last reference
1040   // to |this|.  Hold a reference through this function.
1041   scoped_refptr<SpdySession> self(this);
1042
1043   DCHECK_LT(err, OK);
1044   net_log_.AddEvent(
1045       NetLog::TYPE_SPDY_SESSION_CLOSE,
1046       base::Bind(&NetLogSpdySessionCloseCallback, err, &description));
1047
1048   // Don't close twice.  This can occur because we can have both
1049   // a read and a write outstanding, and each can complete with
1050   // an error.
1051   if (state_ != CLOSED) {
1052     state_ = CLOSED;
1053     error_ = err;
1054     if (remove_from_pool)
1055       RemoveFromPool();
1056     CloseAllStreams(err);
1057   }
1058 }
1059
1060 Value* SpdySession::GetInfoAsValue() const {
1061   DictionaryValue* dict = new DictionaryValue();
1062
1063   dict->SetInteger("source_id", net_log_.source().id);
1064
1065   dict->SetString("host_port_pair", host_port_proxy_pair_.first.ToString());
1066   if (!pooled_aliases_.empty()) {
1067     ListValue* alias_list = new ListValue();
1068     for (std::set<HostPortProxyPair>::const_iterator it =
1069              pooled_aliases_.begin();
1070          it != pooled_aliases_.end(); it++) {
1071       alias_list->Append(Value::CreateStringValue(it->first.ToString()));
1072     }
1073     dict->Set("aliases", alias_list);
1074   }
1075   dict->SetString("proxy", host_port_proxy_pair_.second.ToURI());
1076
1077   dict->SetInteger("active_streams", active_streams_.size());
1078
1079   dict->SetInteger("unclaimed_pushed_streams",
1080       unclaimed_pushed_streams_.size());
1081
1082   dict->SetBoolean("is_secure", is_secure_);
1083
1084   dict->SetString("protocol_negotiated",
1085                   SSLClientSocket::NextProtoToString(
1086                       connection_->socket()->GetNegotiatedProtocol()));
1087
1088   dict->SetInteger("error", error_);
1089   dict->SetInteger("max_concurrent_streams", max_concurrent_streams_);
1090
1091   dict->SetInteger("streams_initiated_count", streams_initiated_count_);
1092   dict->SetInteger("streams_pushed_count", streams_pushed_count_);
1093   dict->SetInteger("streams_pushed_and_claimed_count",
1094       streams_pushed_and_claimed_count_);
1095   dict->SetInteger("streams_abandoned_count", streams_abandoned_count_);
1096   DCHECK(buffered_spdy_framer_.get());
1097   dict->SetInteger("frames_received", buffered_spdy_framer_->frames_received());
1098
1099   dict->SetBoolean("sent_settings", sent_settings_);
1100   dict->SetBoolean("received_settings", received_settings_);
1101   return dict;
1102 }
1103
1104 bool SpdySession::IsReused() const {
1105   return buffered_spdy_framer_->frames_received() > 0;
1106 }
1107
1108 int SpdySession::GetPeerAddress(IPEndPoint* address) const {
1109   if (!connection_->socket())
1110     return ERR_SOCKET_NOT_CONNECTED;
1111
1112   return connection_->socket()->GetPeerAddress(address);
1113 }
1114
1115 int SpdySession::GetLocalAddress(IPEndPoint* address) const {
1116   if (!connection_->socket())
1117     return ERR_SOCKET_NOT_CONNECTED;
1118
1119   return connection_->socket()->GetLocalAddress(address);
1120 }
1121
1122 class SimpleSpdyIOBufferProducer : public SpdySession::SpdyIOBufferProducer {
1123  public:
1124   SimpleSpdyIOBufferProducer(SpdyFrame* frame,
1125                              RequestPriority priority)
1126       : frame_(frame),
1127         priority_(priority) {
1128   }
1129
1130   virtual RequestPriority GetPriority() const OVERRIDE {
1131     return priority_;
1132   }
1133
1134   virtual SpdyIOBuffer* ProduceNextBuffer(SpdySession* session) OVERRIDE {
1135     return SpdySession::SpdyIOBufferProducer::CreateIOBuffer(
1136         frame_.get(), priority_, NULL);
1137   }
1138
1139  private:
1140   scoped_ptr<SpdyFrame> frame_;
1141   RequestPriority priority_;
1142 };
1143
1144 void SpdySession::QueueFrame(SpdyFrame* frame,
1145                              RequestPriority priority) {
1146   SimpleSpdyIOBufferProducer* producer =
1147       new SimpleSpdyIOBufferProducer(frame, priority);
1148   write_queue_.push(producer);
1149   WriteSocketLater();
1150 }
1151
1152 void SpdySession::ActivateStream(SpdyStream* stream) {
1153   if (stream->stream_id() == 0) {
1154     stream->set_stream_id(GetNewStreamId());
1155     created_streams_.erase(scoped_refptr<SpdyStream>(stream));
1156   }
1157   const SpdyStreamId id = stream->stream_id();
1158   DCHECK(!IsStreamActive(id));
1159
1160   active_streams_[id] = stream;
1161 }
1162
1163 void SpdySession::DeleteStream(SpdyStreamId id, int status) {
1164   // For push streams, if they are being deleted normally, we leave
1165   // the stream in the unclaimed_pushed_streams_ list.  However, if
1166   // the stream is errored out, clean it up entirely.
1167   if (status != OK) {
1168     PushedStreamMap::iterator it;
1169     for (it = unclaimed_pushed_streams_.begin();
1170          it != unclaimed_pushed_streams_.end(); ++it) {
1171       scoped_refptr<SpdyStream> curr = it->second.first;
1172       if (id == curr->stream_id()) {
1173         unclaimed_pushed_streams_.erase(it);
1174         break;
1175       }
1176     }
1177   }
1178
1179   // The stream might have been deleted.
1180   ActiveStreamMap::iterator it2 = active_streams_.find(id);
1181   if (it2 == active_streams_.end())
1182     return;
1183
1184   // Possibly remove from the write queue.
1185   WriteQueue old = write_queue_;
1186   write_queue_ = WriteQueue();
1187   while (!old.empty()) {
1188     scoped_ptr<SpdyIOBufferProducer> producer(old.top());
1189     old.pop();
1190     StreamProducerMap::iterator it = stream_producers_.find(producer.get());
1191     if (it == stream_producers_.end() || it->second->stream_id() != id) {
1192       write_queue_.push(producer.release());
1193     } else {
1194       stream_producers_.erase(producer.get());
1195       producer.reset(NULL);
1196     }
1197   }
1198
1199   // If this is an active stream, call the callback.
1200   const scoped_refptr<SpdyStream> stream(it2->second);
1201   active_streams_.erase(it2);
1202   if (stream)
1203     stream->OnClose(status);
1204   ProcessPendingCreateStreams();
1205 }
1206
1207 void SpdySession::RemoveFromPool() {
1208   if (spdy_session_pool_) {
1209     SpdySessionPool* pool = spdy_session_pool_;
1210     spdy_session_pool_ = NULL;
1211     pool->Remove(make_scoped_refptr(this));
1212   }
1213 }
1214
1215 scoped_refptr<SpdyStream> SpdySession::GetActivePushStream(
1216     const std::string& path) {
1217   base::StatsCounter used_push_streams("spdy.claimed_push_streams");
1218
1219   PushedStreamMap::iterator it = unclaimed_pushed_streams_.find(path);
1220   if (it != unclaimed_pushed_streams_.end()) {
1221     net_log_.AddEvent(NetLog::TYPE_SPDY_STREAM_ADOPTED_PUSH_STREAM);
1222     scoped_refptr<SpdyStream> stream = it->second.first;
1223     unclaimed_pushed_streams_.erase(it);
1224     used_push_streams.Increment();
1225     return stream;
1226   }
1227   return NULL;
1228 }
1229
1230 bool SpdySession::GetSSLInfo(SSLInfo* ssl_info,
1231                              bool* was_npn_negotiated,
1232                              NextProto* protocol_negotiated) {
1233
1234   *was_npn_negotiated = connection_->socket()->WasNpnNegotiated();
1235   *protocol_negotiated = connection_->socket()->GetNegotiatedProtocol();
1236   return connection_->socket()->GetSSLInfo(ssl_info);
1237 }
1238
1239 bool SpdySession::GetSSLCertRequestInfo(
1240     SSLCertRequestInfo* cert_request_info) {
1241   if (!is_secure_)
1242     return false;
1243   GetSSLClientSocket()->GetSSLCertRequestInfo(cert_request_info);
1244   return true;
1245 }
1246
1247 ServerBoundCertService* SpdySession::GetServerBoundCertService() const {
1248   if (!is_secure_)
1249     return NULL;
1250   return GetSSLClientSocket()->GetServerBoundCertService();
1251 }
1252
1253 void SpdySession::OnError(SpdyFramer::SpdyError error_code) {
1254   RecordProtocolErrorHistogram(
1255       static_cast<SpdyProtocolErrorDetails>(error_code));
1256   std::string description = base::StringPrintf(
1257       "SPDY_ERROR error_code: %d.", error_code);
1258   CloseSessionOnError(net::ERR_SPDY_PROTOCOL_ERROR, true, description);
1259 }
1260
1261 void SpdySession::OnStreamError(SpdyStreamId stream_id,
1262                                 const std::string& description) {
1263   if (IsStreamActive(stream_id))
1264     ResetStream(stream_id, PROTOCOL_ERROR, description);
1265 }
1266
1267 void SpdySession::OnStreamFrameData(SpdyStreamId stream_id,
1268                                     const char* data,
1269                                     size_t len,
1270                                     SpdyDataFlags flags) {
1271   if (net_log().IsLoggingAllEvents()) {
1272     net_log().AddEvent(
1273         NetLog::TYPE_SPDY_SESSION_RECV_DATA,
1274         base::Bind(&NetLogSpdyDataCallback, stream_id, len, flags));
1275   }
1276
1277   if (!IsStreamActive(stream_id)) {
1278     // NOTE:  it may just be that the stream was cancelled.
1279     return;
1280   }
1281
1282   scoped_refptr<SpdyStream> stream = active_streams_[stream_id];
1283   stream->OnDataReceived(data, len);
1284 }
1285
1286 void SpdySession::OnSetting(SpdySettingsIds id,
1287                             uint8 flags,
1288                             uint32 value) {
1289   HandleSetting(id, value);
1290   http_server_properties_->SetSpdySetting(
1291       host_port_pair(),
1292       id,
1293       static_cast<SpdySettingsFlags>(flags),
1294       value);
1295   received_settings_ = true;
1296
1297   // Log the setting.
1298   net_log_.AddEvent(
1299       NetLog::TYPE_SPDY_SESSION_RECV_SETTING,
1300       base::Bind(&NetLogSpdySettingCallback,
1301                  id, static_cast<SpdySettingsFlags>(flags), value));
1302 }
1303
1304 void SpdySession::OnControlFrameCompressed(
1305     const SpdyControlFrame& uncompressed_frame,
1306     const SpdyControlFrame& compressed_frame) {
1307   if (uncompressed_frame.type() == SYN_STREAM) {
1308     int uncompressed_size = uncompressed_frame.length();
1309     int compressed_size = compressed_frame.length();
1310     // Make sure we avoid early decimal truncation.
1311     int compression_pct = 100 - (100 * compressed_size) / uncompressed_size;
1312     UMA_HISTOGRAM_PERCENTAGE("Net.SpdySynStreamCompressionPercentage",
1313                              compression_pct);
1314   }
1315 }
1316
1317
1318 bool SpdySession::Respond(const SpdyHeaderBlock& headers,
1319                           const scoped_refptr<SpdyStream> stream) {
1320   int rv = OK;
1321   rv = stream->OnResponseReceived(headers);
1322   if (rv < 0) {
1323     DCHECK_NE(rv, ERR_IO_PENDING);
1324     const SpdyStreamId stream_id = stream->stream_id();
1325     DeleteStream(stream_id, rv);
1326     return false;
1327   }
1328   return true;
1329 }
1330
1331 void SpdySession::OnSynStream(SpdyStreamId stream_id,
1332                               SpdyStreamId associated_stream_id,
1333                               SpdyPriority priority,
1334                               uint8 credential_slot,
1335                               bool fin,
1336                               bool unidirectional,
1337                               const SpdyHeaderBlock& headers) {
1338   if (net_log_.IsLoggingAllEvents()) {
1339     net_log_.AddEvent(
1340         NetLog::TYPE_SPDY_SESSION_PUSHED_SYN_STREAM,
1341         base::Bind(&NetLogSpdySynCallback,
1342                    &headers, fin, unidirectional,
1343                    stream_id, associated_stream_id));
1344   }
1345
1346   // Server-initiated streams should have even sequence numbers.
1347   if ((stream_id & 0x1) != 0) {
1348     LOG(WARNING) << "Received invalid OnSyn stream id " << stream_id;
1349     return;
1350   }
1351
1352   if (IsStreamActive(stream_id)) {
1353     LOG(WARNING) << "Received OnSyn for active stream " << stream_id;
1354     return;
1355   }
1356
1357   if (associated_stream_id == 0) {
1358     std::string description = base::StringPrintf(
1359         "Received invalid OnSyn associated stream id %d for stream %d",
1360         associated_stream_id, stream_id);
1361     ResetStream(stream_id, REFUSED_STREAM, description);
1362     return;
1363   }
1364
1365   streams_pushed_count_++;
1366
1367   // TODO(mbelshe): DCHECK that this is a GET method?
1368
1369   // Verify that the response had a URL for us.
1370   GURL gurl = GetUrlFromHeaderBlock(headers, GetProtocolVersion(), true);
1371   if (!gurl.is_valid()) {
1372     ResetStream(stream_id, PROTOCOL_ERROR,
1373                 "Pushed stream url was invalid: " + gurl.spec());
1374     return;
1375   }
1376   const std::string& url = gurl.spec();
1377
1378   // Verify we have a valid stream association.
1379   if (!IsStreamActive(associated_stream_id)) {
1380     ResetStream(stream_id, INVALID_STREAM,
1381                 base::StringPrintf(
1382                     "Received OnSyn with inactive associated stream %d",
1383                     associated_stream_id));
1384     return;
1385   }
1386
1387   // Check that the SYN advertises the same origin as its associated stream.
1388   // Bypass this check if and only if this session is with a SPDY proxy that
1389   // is trusted explicitly via the --trusted-spdy-proxy switch.
1390   if (trusted_spdy_proxy_.Equals(host_port_pair())) {
1391     // Disallow pushing of HTTPS content.
1392     if (gurl.SchemeIs("https")) {
1393       ResetStream(stream_id, REFUSED_STREAM,
1394                   base::StringPrintf(
1395                       "Rejected push of Cross Origin HTTPS content %d",
1396                       associated_stream_id));
1397     }
1398   } else {
1399     scoped_refptr<SpdyStream> associated_stream =
1400         active_streams_[associated_stream_id];
1401     GURL associated_url(associated_stream->GetUrl());
1402     if (associated_url.GetOrigin() != gurl.GetOrigin()) {
1403       ResetStream(stream_id, REFUSED_STREAM,
1404                   base::StringPrintf(
1405                       "Rejected Cross Origin Push Stream %d",
1406                       associated_stream_id));
1407       return;
1408     }
1409   }
1410
1411   // There should not be an existing pushed stream with the same path.
1412   PushedStreamMap::iterator it = unclaimed_pushed_streams_.find(url);
1413   if (it != unclaimed_pushed_streams_.end()) {
1414     ResetStream(stream_id, PROTOCOL_ERROR,
1415                 "Received duplicate pushed stream with url: " + url);
1416     return;
1417   }
1418
1419   scoped_refptr<SpdyStream> stream(new SpdyStream(this, true, net_log_));
1420   stream->set_stream_id(stream_id);
1421
1422   stream->set_path(gurl.PathForRequest());
1423   stream->set_send_window_size(initial_send_window_size_);
1424   stream->set_recv_window_size(initial_recv_window_size_);
1425
1426   DeleteExpiredPushedStreams();
1427   unclaimed_pushed_streams_[url] =
1428       std::pair<scoped_refptr<SpdyStream>, base::TimeTicks> (
1429           stream, time_func_());
1430
1431
1432   ActivateStream(stream);
1433   stream->set_response_received();
1434
1435   // Parse the headers.
1436   if (!Respond(headers, stream))
1437     return;
1438
1439   base::StatsCounter push_requests("spdy.pushed_streams");
1440   push_requests.Increment();
1441 }
1442
1443 void SpdySession::DeleteExpiredPushedStreams() {
1444   if (unclaimed_pushed_streams_.empty())
1445     return;
1446
1447   // Check that adequate time has elapsed since the last sweep.
1448   if (time_func_() < next_unclaimed_push_stream_sweep_time_)
1449     return;
1450
1451   // Delete old streams.
1452   base::TimeTicks minimum_freshness = time_func_() -
1453       base::TimeDelta::FromSeconds(kMinPushedStreamLifetimeSeconds);
1454   PushedStreamMap::iterator it;
1455   for (it = unclaimed_pushed_streams_.begin();
1456       it != unclaimed_pushed_streams_.end(); ) {
1457     const scoped_refptr<SpdyStream>& stream = it->second.first;
1458     base::TimeTicks creation_time = it->second.second;
1459     // DeleteStream() will invalidate the current iterator, so move to next.
1460     ++it;
1461     if (minimum_freshness > creation_time) {
1462       DeleteStream(stream->stream_id(), ERR_INVALID_SPDY_STREAM);
1463       base::StatsCounter abandoned_push_streams(
1464           "spdy.abandoned_push_streams");
1465       base::StatsCounter abandoned_streams("spdy.abandoned_streams");
1466       abandoned_push_streams.Increment();
1467       abandoned_streams.Increment();
1468       streams_abandoned_count_++;
1469     }
1470   }
1471   next_unclaimed_push_stream_sweep_time_ = time_func_() +
1472       base::TimeDelta::FromSeconds(kMinPushedStreamLifetimeSeconds);
1473 }
1474
1475 void SpdySession::OnSynReply(SpdyStreamId stream_id,
1476                              bool fin,
1477                              const SpdyHeaderBlock& headers) {
1478   if (net_log().IsLoggingAllEvents()) {
1479     net_log().AddEvent(
1480         NetLog::TYPE_SPDY_SESSION_SYN_REPLY,
1481         base::Bind(&NetLogSpdySynCallback,
1482                    &headers, fin, false,// not unidirectional
1483                    stream_id, 0));
1484   }
1485
1486   if (!IsStreamActive(stream_id)) {
1487     // NOTE:  it may just be that the stream was cancelled.
1488     LOG(WARNING) << "Received SYN_REPLY for invalid stream " << stream_id;
1489     return;
1490   }
1491
1492   scoped_refptr<SpdyStream> stream = active_streams_[stream_id];
1493   CHECK_EQ(stream->stream_id(), stream_id);
1494   CHECK(!stream->cancelled());
1495
1496   if (stream->response_received()) {
1497     stream->LogStreamError(ERR_SYN_REPLY_NOT_RECEIVED,
1498                            "Received duplicate SYN_REPLY for stream.");
1499     RecordProtocolErrorHistogram(PROTOCOL_ERROR_SYN_REPLY_NOT_RECEIVED);
1500     CloseStream(stream->stream_id(), ERR_SPDY_PROTOCOL_ERROR);
1501     return;
1502   }
1503   stream->set_response_received();
1504
1505   Respond(headers, stream);
1506 }
1507
1508 void SpdySession::OnHeaders(SpdyStreamId stream_id,
1509                             bool fin,
1510                             const SpdyHeaderBlock& headers) {
1511   if (net_log().IsLoggingAllEvents()) {
1512     net_log().AddEvent(
1513         NetLog::TYPE_SPDY_SESSION_RECV_HEADERS,
1514         base::Bind(&NetLogSpdySynCallback,
1515                    &headers, fin, /*unidirectional=*/false,
1516                    stream_id, 0));
1517   }
1518
1519   if (!IsStreamActive(stream_id)) {
1520     // NOTE:  it may just be that the stream was cancelled.
1521     LOG(WARNING) << "Received HEADERS for invalid stream " << stream_id;
1522     return;
1523   }
1524
1525   scoped_refptr<SpdyStream> stream = active_streams_[stream_id];
1526   CHECK_EQ(stream->stream_id(), stream_id);
1527   CHECK(!stream->cancelled());
1528
1529   int rv = stream->OnHeaders(headers);
1530   if (rv < 0) {
1531     DCHECK_NE(rv, ERR_IO_PENDING);
1532     const SpdyStreamId stream_id = stream->stream_id();
1533     DeleteStream(stream_id, rv);
1534   }
1535 }
1536
1537 void SpdySession::OnRstStream(SpdyStreamId stream_id, SpdyStatusCodes status) {
1538   std::string description;
1539   net_log().AddEvent(
1540       NetLog::TYPE_SPDY_SESSION_RST_STREAM,
1541       base::Bind(&NetLogSpdyRstCallback,
1542                  stream_id, status, &description));
1543
1544   if (!IsStreamActive(stream_id)) {
1545     // NOTE:  it may just be that the stream was cancelled.
1546     LOG(WARNING) << "Received RST for invalid stream" << stream_id;
1547     return;
1548   }
1549   scoped_refptr<SpdyStream> stream = active_streams_[stream_id];
1550   CHECK_EQ(stream->stream_id(), stream_id);
1551   CHECK(!stream->cancelled());
1552
1553   if (status == 0) {
1554     stream->OnDataReceived(NULL, 0);
1555   } else if (status == REFUSED_STREAM) {
1556     DeleteStream(stream_id, ERR_SPDY_SERVER_REFUSED_STREAM);
1557   } else {
1558     RecordProtocolErrorHistogram(
1559         PROTOCOL_ERROR_RST_STREAM_FOR_NON_ACTIVE_STREAM);
1560     stream->LogStreamError(ERR_SPDY_PROTOCOL_ERROR,
1561                            base::StringPrintf("SPDY stream closed: %d",
1562                                               status));
1563     // TODO(mbelshe): Map from Spdy-protocol errors to something sensical.
1564     //                For now, it doesn't matter much - it is a protocol error.
1565     DeleteStream(stream_id, ERR_SPDY_PROTOCOL_ERROR);
1566   }
1567 }
1568
1569 void SpdySession::OnGoAway(SpdyStreamId last_accepted_stream_id,
1570                            SpdyGoAwayStatus status) {
1571   net_log_.AddEvent(NetLog::TYPE_SPDY_SESSION_GOAWAY,
1572       base::Bind(&NetLogSpdyGoAwayCallback,
1573                  last_accepted_stream_id,
1574                  active_streams_.size(),
1575                  unclaimed_pushed_streams_.size(),
1576                  status));
1577   RemoveFromPool();
1578   CloseAllStreams(net::ERR_ABORTED);
1579
1580   // TODO(willchan): Cancel any streams that are past the GoAway frame's
1581   // |last_accepted_stream_id|.
1582
1583   // Don't bother killing any streams that are still reading.  They'll either
1584   // complete successfully or get an ERR_CONNECTION_CLOSED when the socket is
1585   // closed.
1586 }
1587
1588 void SpdySession::OnPing(uint32 unique_id) {
1589   net_log_.AddEvent(
1590       NetLog::TYPE_SPDY_SESSION_PING,
1591       base::Bind(&NetLogSpdyPingCallback, unique_id, "received"));
1592
1593   // Send response to a PING from server.
1594   if (unique_id % 2 == 0) {
1595     WritePingFrame(unique_id);
1596     return;
1597   }
1598
1599   --pings_in_flight_;
1600   if (pings_in_flight_ < 0) {
1601     RecordProtocolErrorHistogram(PROTOCOL_ERROR_UNEXPECTED_PING);
1602     CloseSessionOnError(
1603         net::ERR_SPDY_PROTOCOL_ERROR, true, "pings_in_flight_ is < 0.");
1604     pings_in_flight_ = 0;
1605     return;
1606   }
1607
1608   if (pings_in_flight_ > 0)
1609     return;
1610
1611   // We will record RTT in histogram when there are no more client sent
1612   // pings_in_flight_.
1613   RecordPingRTTHistogram(base::TimeTicks::Now() - last_ping_sent_time_);
1614 }
1615
1616 void SpdySession::OnWindowUpdate(SpdyStreamId stream_id,
1617                                  int delta_window_size) {
1618   net_log_.AddEvent(
1619       NetLog::TYPE_SPDY_SESSION_RECEIVED_WINDOW_UPDATE,
1620       base::Bind(&NetLogSpdyWindowUpdateCallback,
1621                  stream_id, delta_window_size));
1622
1623   if (!IsStreamActive(stream_id)) {
1624     LOG(WARNING) << "Received WINDOW_UPDATE for invalid stream " << stream_id;
1625     return;
1626   }
1627
1628   if (delta_window_size < 1) {
1629     ResetStream(stream_id, FLOW_CONTROL_ERROR,
1630                 base::StringPrintf(
1631                     "Received WINDOW_UPDATE with an invalid "
1632                     "delta_window_size %d", delta_window_size));
1633     return;
1634   }
1635
1636   scoped_refptr<SpdyStream> stream = active_streams_[stream_id];
1637   CHECK_EQ(stream->stream_id(), stream_id);
1638   CHECK(!stream->cancelled());
1639
1640   if (flow_control_)
1641     stream->IncreaseSendWindowSize(delta_window_size);
1642 }
1643
1644 void SpdySession::SendWindowUpdate(SpdyStreamId stream_id,
1645                                    int32 delta_window_size) {
1646   CHECK(IsStreamActive(stream_id));
1647   scoped_refptr<SpdyStream> stream = active_streams_[stream_id];
1648   CHECK_EQ(stream->stream_id(), stream_id);
1649
1650   net_log_.AddEvent(
1651       NetLog::TYPE_SPDY_SESSION_SENT_WINDOW_UPDATE,
1652       base::Bind(&NetLogSpdyWindowUpdateCallback,
1653                  stream_id, delta_window_size));
1654
1655   DCHECK(buffered_spdy_framer_.get());
1656   scoped_ptr<SpdyWindowUpdateControlFrame> window_update_frame(
1657       buffered_spdy_framer_->CreateWindowUpdate(stream_id, delta_window_size));
1658   QueueFrame(window_update_frame.release(), stream->priority());
1659 }
1660
1661 // Given a cwnd that we would have sent to the server, modify it based on the
1662 // field trial policy.
1663 uint32 ApplyCwndFieldTrialPolicy(int cwnd) {
1664   base::FieldTrial* trial = base::FieldTrialList::Find("SpdyCwnd");
1665   if (!trial) {
1666       LOG(WARNING) << "Could not find \"SpdyCwnd\" in FieldTrialList";
1667       return cwnd;
1668   }
1669   if (trial->group_name() == "cwnd10")
1670     return 10;
1671   else if (trial->group_name() == "cwnd16")
1672     return 16;
1673   else if (trial->group_name() == "cwndMin16")
1674     return std::max(cwnd, 16);
1675   else if (trial->group_name() == "cwndMin10")
1676     return std::max(cwnd, 10);
1677   else if (trial->group_name() == "cwndDynamic")
1678     return cwnd;
1679   NOTREACHED();
1680   return cwnd;
1681 }
1682
1683 void SpdySession::SendInitialSettings() {
1684   // First notify the server about the settings they should use when
1685   // communicating with us.
1686   if (GetProtocolVersion() >= 2 && enable_sending_initial_settings_) {
1687     SettingsMap settings_map;
1688     // Create a new settings frame notifying the sever of our
1689     // max_concurrent_streams_ and initial window size.
1690     settings_map[SETTINGS_MAX_CONCURRENT_STREAMS] =
1691         SettingsFlagsAndValue(SETTINGS_FLAG_NONE, kMaxConcurrentPushedStreams);
1692     if (GetProtocolVersion() > 2 &&
1693         initial_recv_window_size_ != kSpdyStreamInitialWindowSize) {
1694       settings_map[SETTINGS_INITIAL_WINDOW_SIZE] =
1695           SettingsFlagsAndValue(SETTINGS_FLAG_NONE, initial_recv_window_size_);
1696     }
1697     SendSettings(settings_map);
1698   }
1699
1700   // Next notify the server about the settings they have previously
1701   // told us to use when communicating with them.
1702   const SettingsMap& settings_map =
1703       http_server_properties_->GetSpdySettings(host_port_pair());
1704   if (settings_map.empty())
1705     return;
1706
1707   // Record Histogram Data and Apply the SpdyCwnd FieldTrial if applicable.
1708   const SpdySettingsIds id = SETTINGS_CURRENT_CWND;
1709   SettingsMap::const_iterator it = settings_map.find(id);
1710   uint32 value = 0;
1711   if (it != settings_map.end())
1712     value = it->second.second;
1713   uint32 cwnd = ApplyCwndFieldTrialPolicy(value);
1714   UMA_HISTOGRAM_CUSTOM_COUNTS("Net.SpdySettingsCwndSent", cwnd, 1, 200, 100);
1715   if (cwnd != value) {
1716     http_server_properties_->SetSpdySetting(
1717         host_port_pair(), id, SETTINGS_FLAG_PLEASE_PERSIST, cwnd);
1718   }
1719
1720   const SettingsMap& settings_map_new =
1721       http_server_properties_->GetSpdySettings(host_port_pair());
1722   for (SettingsMap::const_iterator i = settings_map_new.begin(),
1723            end = settings_map_new.end(); i != end; ++i) {
1724     const SpdySettingsIds new_id = i->first;
1725     const uint32 new_val = i->second.second;
1726     HandleSetting(new_id, new_val);
1727   }
1728
1729   SendSettings(settings_map_new);
1730 }
1731
1732
1733 void SpdySession::SendSettings(const SettingsMap& settings) {
1734   net_log_.AddEvent(
1735       NetLog::TYPE_SPDY_SESSION_SEND_SETTINGS,
1736       base::Bind(&NetLogSpdySettingsCallback, &settings));
1737
1738   // Create the SETTINGS frame and send it.
1739   DCHECK(buffered_spdy_framer_.get());
1740   scoped_ptr<SpdySettingsControlFrame> settings_frame(
1741       buffered_spdy_framer_->CreateSettings(settings));
1742   sent_settings_ = true;
1743   QueueFrame(settings_frame.release(), HIGHEST);
1744 }
1745
1746 void SpdySession::HandleSetting(uint32 id, uint32 value) {
1747   switch (id) {
1748     case SETTINGS_MAX_CONCURRENT_STREAMS:
1749       max_concurrent_streams_ = std::min(static_cast<size_t>(value),
1750                                          kMaxConcurrentStreamLimit);
1751       ProcessPendingCreateStreams();
1752       break;
1753     case SETTINGS_INITIAL_WINDOW_SIZE:
1754       if (static_cast<int32>(value) < 0) {
1755         net_log().AddEvent(
1756             NetLog::TYPE_SPDY_SESSION_NEGATIVE_INITIAL_WINDOW_SIZE,
1757             NetLog::IntegerCallback("initial_window_size", value));
1758       } else {
1759         // SETTINGS_INITIAL_WINDOW_SIZE updates initial_send_window_size_ only.
1760         int32 delta_window_size = value - initial_send_window_size_;
1761         initial_send_window_size_ = value;
1762         UpdateStreamsSendWindowSize(delta_window_size);
1763         net_log().AddEvent(
1764             NetLog::TYPE_SPDY_SESSION_UPDATE_STREAMS_SEND_WINDOW_SIZE,
1765             NetLog::IntegerCallback("delta_window_size", delta_window_size));
1766       }
1767       break;
1768   }
1769 }
1770
1771 void SpdySession::UpdateStreamsSendWindowSize(int32 delta_window_size) {
1772   ActiveStreamMap::iterator it;
1773   for (it = active_streams_.begin(); it != active_streams_.end(); ++it) {
1774     const scoped_refptr<SpdyStream>& stream = it->second;
1775     DCHECK(stream);
1776     stream->AdjustSendWindowSize(delta_window_size);
1777   }
1778
1779   CreatedStreamSet::iterator i;
1780   for (i = created_streams_.begin(); i != created_streams_.end(); i++) {
1781     const scoped_refptr<SpdyStream>& stream = *i;
1782     stream->AdjustSendWindowSize(delta_window_size);
1783   }
1784 }
1785
1786 void SpdySession::SendPrefacePingIfNoneInFlight() {
1787   if (pings_in_flight_ || !enable_ping_based_connection_checking_)
1788     return;
1789
1790   base::TimeTicks now = base::TimeTicks::Now();
1791   // If there is no activity in the session, then send a preface-PING.
1792   if ((now - last_activity_time_) > connection_at_risk_of_loss_time_)
1793     SendPrefacePing();
1794 }
1795
1796 void SpdySession::SendPrefacePing() {
1797   WritePingFrame(next_ping_id_);
1798 }
1799
1800 void SpdySession::WritePingFrame(uint32 unique_id) {
1801   DCHECK(buffered_spdy_framer_.get());
1802   scoped_ptr<SpdyPingControlFrame> ping_frame(
1803       buffered_spdy_framer_->CreatePingFrame(unique_id));
1804   QueueFrame(ping_frame.release(), HIGHEST);
1805
1806   if (net_log().IsLoggingAllEvents()) {
1807     net_log().AddEvent(
1808         NetLog::TYPE_SPDY_SESSION_PING,
1809         base::Bind(&NetLogSpdyPingCallback, unique_id, "sent"));
1810   }
1811   if (unique_id % 2 != 0) {
1812     next_ping_id_ += 2;
1813     ++pings_in_flight_;
1814     PlanToCheckPingStatus();
1815     last_ping_sent_time_ = base::TimeTicks::Now();
1816   }
1817 }
1818
1819 void SpdySession::PlanToCheckPingStatus() {
1820   if (check_ping_status_pending_)
1821     return;
1822
1823   check_ping_status_pending_ = true;
1824   MessageLoop::current()->PostDelayedTask(
1825       FROM_HERE,
1826       base::Bind(&SpdySession::CheckPingStatus, weak_factory_.GetWeakPtr(),
1827                  base::TimeTicks::Now()), hung_interval_);
1828 }
1829
1830 void SpdySession::CheckPingStatus(base::TimeTicks last_check_time) {
1831   // Check if we got a response back for all PINGs we had sent.
1832   if (pings_in_flight_ == 0) {
1833     check_ping_status_pending_ = false;
1834     return;
1835   }
1836
1837   DCHECK(check_ping_status_pending_);
1838
1839   base::TimeTicks now = base::TimeTicks::Now();
1840   base::TimeDelta delay = hung_interval_ - (now - last_activity_time_);
1841
1842   if (delay.InMilliseconds() < 0 || last_activity_time_ < last_check_time) {
1843     CloseSessionOnError(net::ERR_SPDY_PING_FAILED, true, "Failed ping.");
1844     // Track all failed PING messages in a separate bucket.
1845     const base::TimeDelta kFailedPing =
1846         base::TimeDelta::FromInternalValue(INT_MAX);
1847     RecordPingRTTHistogram(kFailedPing);
1848     return;
1849   }
1850
1851   // Check the status of connection after a delay.
1852   MessageLoop::current()->PostDelayedTask(
1853       FROM_HERE,
1854       base::Bind(&SpdySession::CheckPingStatus, weak_factory_.GetWeakPtr(),
1855                  now),
1856       delay);
1857 }
1858
1859 void SpdySession::RecordPingRTTHistogram(base::TimeDelta duration) {
1860   UMA_HISTOGRAM_TIMES("Net.SpdyPing.RTT", duration);
1861 }
1862
1863 void SpdySession::RecordProtocolErrorHistogram(
1864     SpdyProtocolErrorDetails details) {
1865   UMA_HISTOGRAM_ENUMERATION("Net.SpdySessionErrorDetails", details,
1866                             NUM_SPDY_PROTOCOL_ERROR_DETAILS);
1867   if (EndsWith(host_port_pair().host(), "google.com", false)) {
1868     UMA_HISTOGRAM_ENUMERATION("Net.SpdySessionErrorDetails_Google", details,
1869                               NUM_SPDY_PROTOCOL_ERROR_DETAILS);
1870   }
1871 }
1872
1873 void SpdySession::RecordHistograms() {
1874   UMA_HISTOGRAM_CUSTOM_COUNTS("Net.SpdyStreamsPerSession",
1875                               streams_initiated_count_,
1876                               0, 300, 50);
1877   UMA_HISTOGRAM_CUSTOM_COUNTS("Net.SpdyStreamsPushedPerSession",
1878                               streams_pushed_count_,
1879                               0, 300, 50);
1880   UMA_HISTOGRAM_CUSTOM_COUNTS("Net.SpdyStreamsPushedAndClaimedPerSession",
1881                               streams_pushed_and_claimed_count_,
1882                               0, 300, 50);
1883   UMA_HISTOGRAM_CUSTOM_COUNTS("Net.SpdyStreamsAbandonedPerSession",
1884                               streams_abandoned_count_,
1885                               0, 300, 50);
1886   UMA_HISTOGRAM_ENUMERATION("Net.SpdySettingsSent",
1887                             sent_settings_ ? 1 : 0, 2);
1888   UMA_HISTOGRAM_ENUMERATION("Net.SpdySettingsReceived",
1889                             received_settings_ ? 1 : 0, 2);
1890   UMA_HISTOGRAM_CUSTOM_COUNTS("Net.SpdyStreamStallsPerSession",
1891                               stalled_streams_,
1892                               0, 300, 50);
1893   UMA_HISTOGRAM_ENUMERATION("Net.SpdySessionsWithStalls",
1894                             stalled_streams_ > 0 ? 1 : 0, 2);
1895
1896   if (received_settings_) {
1897     // Enumerate the saved settings, and set histograms for it.
1898     const SettingsMap& settings_map =
1899         http_server_properties_->GetSpdySettings(host_port_pair());
1900
1901     SettingsMap::const_iterator it;
1902     for (it = settings_map.begin(); it != settings_map.end(); ++it) {
1903       const SpdySettingsIds id = it->first;
1904       const uint32 val = it->second.second;
1905       switch (id) {
1906         case SETTINGS_CURRENT_CWND:
1907           // Record several different histograms to see if cwnd converges
1908           // for larger volumes of data being sent.
1909           UMA_HISTOGRAM_CUSTOM_COUNTS("Net.SpdySettingsCwnd",
1910                                       val, 1, 200, 100);
1911           if (bytes_received_ > 10 * 1024) {
1912             UMA_HISTOGRAM_CUSTOM_COUNTS("Net.SpdySettingsCwnd10K",
1913                                         val, 1, 200, 100);
1914             if (bytes_received_ > 25 * 1024) {
1915               UMA_HISTOGRAM_CUSTOM_COUNTS("Net.SpdySettingsCwnd25K",
1916                                           val, 1, 200, 100);
1917               if (bytes_received_ > 50 * 1024) {
1918                 UMA_HISTOGRAM_CUSTOM_COUNTS("Net.SpdySettingsCwnd50K",
1919                                             val, 1, 200, 100);
1920                 if (bytes_received_ > 100 * 1024) {
1921                   UMA_HISTOGRAM_CUSTOM_COUNTS("Net.SpdySettingsCwnd100K",
1922                                               val, 1, 200, 100);
1923                 }
1924               }
1925             }
1926           }
1927           break;
1928         case SETTINGS_ROUND_TRIP_TIME:
1929           UMA_HISTOGRAM_CUSTOM_COUNTS("Net.SpdySettingsRTT",
1930                                       val, 1, 1200, 100);
1931           break;
1932         case SETTINGS_DOWNLOAD_RETRANS_RATE:
1933           UMA_HISTOGRAM_CUSTOM_COUNTS("Net.SpdySettingsRetransRate",
1934                                       val, 1, 100, 50);
1935           break;
1936         default:
1937           break;
1938       }
1939     }
1940   }
1941 }
1942
1943 void SpdySession::InvokeUserStreamCreationCallback(
1944     scoped_refptr<SpdyStream>* stream) {
1945   PendingCallbackMap::iterator it = pending_callback_map_.find(stream);
1946
1947   // Exit if the request has already been cancelled.
1948   if (it == pending_callback_map_.end())
1949     return;
1950
1951   CompletionCallback callback = it->second.callback;
1952   int result = it->second.result;
1953   pending_callback_map_.erase(it);
1954   callback.Run(result);
1955 }
1956
1957 SSLClientSocket* SpdySession::GetSSLClientSocket() const {
1958   if (!is_secure_)
1959     return NULL;
1960   SSLClientSocket* ssl_socket =
1961       reinterpret_cast<SSLClientSocket*>(connection_->socket());
1962   DCHECK(ssl_socket);
1963   return ssl_socket;
1964 }
1965
1966 }  // namespace net