Chromium Code Reviews
chromiumcodereview-hr@appspot.gserviceaccount.com (chromiumcodereview-hr) | Please choose your nickname with Settings | Help | Chromium Project | Gerrit Changes | Sign out
(483)

Issues Search
    My Issues | Starred     Open | Closed | All

Side by Side Diff: net/spdy/spdy_session.cc

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

Powered by Google App Engine
This is Rietveld 408576698