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