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/quic/quic_session.h"
6
7 #include "base/stl_util.h"
8 #include "net/quic/crypto/proof_verifier.h"
9 #include "net/quic/quic_connection.h"
10 #include "net/quic/quic_flags.h"
11 #include "net/quic/quic_flow_controller.h"
12 #include "net/quic/quic_headers_stream.h"
13 #include "net/ssl/ssl_info.h"
14
15 using base::StringPiece;
16 using base::hash_map;
17 using base::hash_set;
18 using std::make_pair;
19 using std::vector;
20
21 namespace net {
22
23 #define ENDPOINT (is_server() ? "Server: " : " Client: ")
24
25 // We want to make sure we delete any closed streams in a safe manner.
26 // To avoid deleting a stream in mid-operation, we have a simple shim between
27 // us and the stream, so we can delete any streams when we return from
28 // processing.
29 //
30 // We could just override the base methods, but this makes it easier to make
31 // sure we don't miss any.
32 class VisitorShim : public QuicConnectionVisitorInterface {
33 public:
VisitorShim(QuicSession * session)34 explicit VisitorShim(QuicSession* session) : session_(session) {}
35
OnStreamFrames(const vector<QuicStreamFrame> & frames)36 virtual void OnStreamFrames(const vector<QuicStreamFrame>& frames) OVERRIDE {
37 session_->OnStreamFrames(frames);
38 session_->PostProcessAfterData();
39 }
OnRstStream(const QuicRstStreamFrame & frame)40 virtual void OnRstStream(const QuicRstStreamFrame& frame) OVERRIDE {
41 session_->OnRstStream(frame);
42 session_->PostProcessAfterData();
43 }
44
OnGoAway(const QuicGoAwayFrame & frame)45 virtual void OnGoAway(const QuicGoAwayFrame& frame) OVERRIDE {
46 session_->OnGoAway(frame);
47 session_->PostProcessAfterData();
48 }
49
OnWindowUpdateFrames(const vector<QuicWindowUpdateFrame> & frames)50 virtual void OnWindowUpdateFrames(const vector<QuicWindowUpdateFrame>& frames)
51 OVERRIDE {
52 session_->OnWindowUpdateFrames(frames);
53 session_->PostProcessAfterData();
54 }
55
OnBlockedFrames(const vector<QuicBlockedFrame> & frames)56 virtual void OnBlockedFrames(const vector<QuicBlockedFrame>& frames)
57 OVERRIDE {
58 session_->OnBlockedFrames(frames);
59 session_->PostProcessAfterData();
60 }
61
OnCanWrite()62 virtual void OnCanWrite() OVERRIDE {
63 session_->OnCanWrite();
64 session_->PostProcessAfterData();
65 }
66
OnSuccessfulVersionNegotiation(const QuicVersion & version)67 virtual void OnSuccessfulVersionNegotiation(
68 const QuicVersion& version) OVERRIDE {
69 session_->OnSuccessfulVersionNegotiation(version);
70 }
71
OnConnectionClosed(QuicErrorCode error,bool from_peer)72 virtual void OnConnectionClosed(
73 QuicErrorCode error, bool from_peer) OVERRIDE {
74 session_->OnConnectionClosed(error, from_peer);
75 // The session will go away, so don't bother with cleanup.
76 }
77
OnWriteBlocked()78 virtual void OnWriteBlocked() OVERRIDE {
79 session_->OnWriteBlocked();
80 }
81
WillingAndAbleToWrite() const82 virtual bool WillingAndAbleToWrite() const OVERRIDE {
83 return session_->WillingAndAbleToWrite();
84 }
85
HasPendingHandshake() const86 virtual bool HasPendingHandshake() const OVERRIDE {
87 return session_->HasPendingHandshake();
88 }
89
HasOpenDataStreams() const90 virtual bool HasOpenDataStreams() const OVERRIDE {
91 return session_->HasOpenDataStreams();
92 }
93
94 private:
95 QuicSession* session_;
96 };
97
QuicSession(QuicConnection * connection,const QuicConfig & config)98 QuicSession::QuicSession(QuicConnection* connection, const QuicConfig& config)
99 : connection_(connection),
100 visitor_shim_(new VisitorShim(this)),
101 config_(config),
102 max_open_streams_(config_.max_streams_per_connection()),
103 next_stream_id_(is_server() ? 2 : 3),
104 largest_peer_created_stream_id_(0),
105 error_(QUIC_NO_ERROR),
106 goaway_received_(false),
107 goaway_sent_(false),
108 has_pending_handshake_(false) {
109 if (connection_->version() <= QUIC_VERSION_19) {
110 flow_controller_.reset(new QuicFlowController(
111 connection_.get(), 0, is_server(), kDefaultFlowControlSendWindow,
112 config_.GetInitialFlowControlWindowToSend(),
113 config_.GetInitialFlowControlWindowToSend()));
114 } else {
115 flow_controller_.reset(new QuicFlowController(
116 connection_.get(), 0, is_server(), kDefaultFlowControlSendWindow,
117 config_.GetInitialSessionFlowControlWindowToSend(),
118 config_.GetInitialSessionFlowControlWindowToSend()));
119 }
120
121 connection_->set_visitor(visitor_shim_.get());
122 connection_->SetFromConfig(config_);
123 if (connection_->connected()) {
124 connection_->SetOverallConnectionTimeout(
125 config_.max_time_before_crypto_handshake());
126 }
127 headers_stream_.reset(new QuicHeadersStream(this));
128 if (!is_server()) {
129 // For version above QUIC v12, the headers stream is stream 3, so the
130 // next available local stream ID should be 5.
131 DCHECK_EQ(kHeadersStreamId, next_stream_id_);
132 next_stream_id_ += 2;
133 }
134 }
135
~QuicSession()136 QuicSession::~QuicSession() {
137 STLDeleteElements(&closed_streams_);
138 STLDeleteValues(&stream_map_);
139
140 DLOG_IF(WARNING,
141 locally_closed_streams_highest_offset_.size() > max_open_streams_)
142 << "Surprisingly high number of locally closed streams still waiting for "
143 "final byte offset: " << locally_closed_streams_highest_offset_.size();
144 }
145
OnStreamFrames(const vector<QuicStreamFrame> & frames)146 void QuicSession::OnStreamFrames(const vector<QuicStreamFrame>& frames) {
147 for (size_t i = 0; i < frames.size(); ++i) {
148 // TODO(rch) deal with the error case of stream id 0.
149 const QuicStreamFrame& frame = frames[i];
150 QuicStreamId stream_id = frame.stream_id;
151 ReliableQuicStream* stream = GetStream(stream_id);
152 if (!stream) {
153 // The stream no longer exists, but we may still be interested in the
154 // final stream byte offset sent by the peer. A frame with a FIN can give
155 // us this offset.
156 if (frame.fin) {
157 QuicStreamOffset final_byte_offset =
158 frame.offset + frame.data.TotalBufferSize();
159 UpdateFlowControlOnFinalReceivedByteOffset(stream_id,
160 final_byte_offset);
161 }
162
163 continue;
164 }
165 stream->OnStreamFrame(frames[i]);
166 }
167 }
168
OnStreamHeaders(QuicStreamId stream_id,StringPiece headers_data)169 void QuicSession::OnStreamHeaders(QuicStreamId stream_id,
170 StringPiece headers_data) {
171 QuicDataStream* stream = GetDataStream(stream_id);
172 if (!stream) {
173 // It's quite possible to receive headers after a stream has been reset.
174 return;
175 }
176 stream->OnStreamHeaders(headers_data);
177 }
178
OnStreamHeadersPriority(QuicStreamId stream_id,QuicPriority priority)179 void QuicSession::OnStreamHeadersPriority(QuicStreamId stream_id,
180 QuicPriority priority) {
181 QuicDataStream* stream = GetDataStream(stream_id);
182 if (!stream) {
183 // It's quite possible to receive headers after a stream has been reset.
184 return;
185 }
186 stream->OnStreamHeadersPriority(priority);
187 }
188
OnStreamHeadersComplete(QuicStreamId stream_id,bool fin,size_t frame_len)189 void QuicSession::OnStreamHeadersComplete(QuicStreamId stream_id,
190 bool fin,
191 size_t frame_len) {
192 QuicDataStream* stream = GetDataStream(stream_id);
193 if (!stream) {
194 // It's quite possible to receive headers after a stream has been reset.
195 return;
196 }
197 stream->OnStreamHeadersComplete(fin, frame_len);
198 }
199
OnRstStream(const QuicRstStreamFrame & frame)200 void QuicSession::OnRstStream(const QuicRstStreamFrame& frame) {
201 if (frame.stream_id == kCryptoStreamId) {
202 connection()->SendConnectionCloseWithDetails(
203 QUIC_INVALID_STREAM_ID,
204 "Attempt to reset the crypto stream");
205 return;
206 }
207 if (frame.stream_id == kHeadersStreamId) {
208 connection()->SendConnectionCloseWithDetails(
209 QUIC_INVALID_STREAM_ID,
210 "Attempt to reset the headers stream");
211 return;
212 }
213
214 QuicDataStream* stream = GetDataStream(frame.stream_id);
215 if (!stream) {
216 // The RST frame contains the final byte offset for the stream: we can now
217 // update the connection level flow controller if needed.
218 UpdateFlowControlOnFinalReceivedByteOffset(frame.stream_id,
219 frame.byte_offset);
220 return; // Errors are handled by GetStream.
221 }
222
223 stream->OnStreamReset(frame);
224 }
225
OnGoAway(const QuicGoAwayFrame & frame)226 void QuicSession::OnGoAway(const QuicGoAwayFrame& frame) {
227 DCHECK(frame.last_good_stream_id < next_stream_id_);
228 goaway_received_ = true;
229 }
230
OnConnectionClosed(QuicErrorCode error,bool from_peer)231 void QuicSession::OnConnectionClosed(QuicErrorCode error, bool from_peer) {
232 DCHECK(!connection_->connected());
233 if (error_ == QUIC_NO_ERROR) {
234 error_ = error;
235 }
236
237 while (!stream_map_.empty()) {
238 DataStreamMap::iterator it = stream_map_.begin();
239 QuicStreamId id = it->first;
240 it->second->OnConnectionClosed(error, from_peer);
241 // The stream should call CloseStream as part of OnConnectionClosed.
242 if (stream_map_.find(id) != stream_map_.end()) {
243 LOG(DFATAL) << ENDPOINT
244 << "Stream failed to close under OnConnectionClosed";
245 CloseStream(id);
246 }
247 }
248 }
249
OnWindowUpdateFrames(const vector<QuicWindowUpdateFrame> & frames)250 void QuicSession::OnWindowUpdateFrames(
251 const vector<QuicWindowUpdateFrame>& frames) {
252 bool connection_window_updated = false;
253 for (size_t i = 0; i < frames.size(); ++i) {
254 // Stream may be closed by the time we receive a WINDOW_UPDATE, so we can't
255 // assume that it still exists.
256 QuicStreamId stream_id = frames[i].stream_id;
257 if (stream_id == 0) {
258 // This is a window update that applies to the connection, rather than an
259 // individual stream.
260 DVLOG(1) << ENDPOINT
261 << "Received connection level flow control window update with "
262 "byte offset: " << frames[i].byte_offset;
263 if (FLAGS_enable_quic_connection_flow_control_2 &&
264 flow_controller_->UpdateSendWindowOffset(frames[i].byte_offset)) {
265 connection_window_updated = true;
266 }
267 continue;
268 }
269
270 QuicDataStream* stream = GetDataStream(stream_id);
271 if (stream) {
272 stream->OnWindowUpdateFrame(frames[i]);
273 }
274 }
275
276 // Connection level flow control window has increased, so blocked streams can
277 // write again.
278 if (connection_window_updated) {
279 OnCanWrite();
280 }
281 }
282
OnBlockedFrames(const vector<QuicBlockedFrame> & frames)283 void QuicSession::OnBlockedFrames(const vector<QuicBlockedFrame>& frames) {
284 for (size_t i = 0; i < frames.size(); ++i) {
285 // TODO(rjshade): Compare our flow control receive windows for specified
286 // streams: if we have a large window then maybe something
287 // had gone wrong with the flow control accounting.
288 DVLOG(1) << ENDPOINT << "Received BLOCKED frame with stream id: "
289 << frames[i].stream_id;
290 }
291 }
292
OnCanWrite()293 void QuicSession::OnCanWrite() {
294 // We limit the number of writes to the number of pending streams. If more
295 // streams become pending, WillingAndAbleToWrite will be true, which will
296 // cause the connection to request resumption before yielding to other
297 // connections.
298 size_t num_writes = write_blocked_streams_.NumBlockedStreams();
299 if (flow_controller_->IsBlocked()) {
300 // If we are connection level flow control blocked, then only allow the
301 // crypto and headers streams to try writing as all other streams will be
302 // blocked.
303 num_writes = 0;
304 if (write_blocked_streams_.crypto_stream_blocked()) {
305 num_writes += 1;
306 }
307 if (write_blocked_streams_.headers_stream_blocked()) {
308 num_writes += 1;
309 }
310 }
311 if (num_writes == 0) {
312 return;
313 }
314
315 QuicConnection::ScopedPacketBundler ack_bundler(
316 connection_.get(), QuicConnection::NO_ACK);
317 for (size_t i = 0; i < num_writes; ++i) {
318 if (!(write_blocked_streams_.HasWriteBlockedCryptoOrHeadersStream() ||
319 write_blocked_streams_.HasWriteBlockedDataStreams())) {
320 // Writing one stream removed another!? Something's broken.
321 LOG(DFATAL) << "WriteBlockedStream is missing";
322 connection_->CloseConnection(QUIC_INTERNAL_ERROR, false);
323 return;
324 }
325 if (!connection_->CanWriteStreamData()) {
326 return;
327 }
328 QuicStreamId stream_id = write_blocked_streams_.PopFront();
329 if (stream_id == kCryptoStreamId) {
330 has_pending_handshake_ = false; // We just popped it.
331 }
332 ReliableQuicStream* stream = GetStream(stream_id);
333 if (stream != NULL && !stream->flow_controller()->IsBlocked()) {
334 // If the stream can't write all bytes, it'll re-add itself to the blocked
335 // list.
336 stream->OnCanWrite();
337 }
338 }
339 }
340
WillingAndAbleToWrite() const341 bool QuicSession::WillingAndAbleToWrite() const {
342 // If the crypto or headers streams are blocked, we want to schedule a write -
343 // they don't get blocked by connection level flow control. Otherwise only
344 // schedule a write if we are not flow control blocked at the connection
345 // level.
346 return write_blocked_streams_.HasWriteBlockedCryptoOrHeadersStream() ||
347 (!flow_controller_->IsBlocked() &&
348 write_blocked_streams_.HasWriteBlockedDataStreams());
349 }
350
HasPendingHandshake() const351 bool QuicSession::HasPendingHandshake() const {
352 return has_pending_handshake_;
353 }
354
HasOpenDataStreams() const355 bool QuicSession::HasOpenDataStreams() const {
356 return GetNumOpenStreams() > 0;
357 }
358
WritevData(QuicStreamId id,const IOVector & data,QuicStreamOffset offset,bool fin,FecProtection fec_protection,QuicAckNotifier::DelegateInterface * ack_notifier_delegate)359 QuicConsumedData QuicSession::WritevData(
360 QuicStreamId id,
361 const IOVector& data,
362 QuicStreamOffset offset,
363 bool fin,
364 FecProtection fec_protection,
365 QuicAckNotifier::DelegateInterface* ack_notifier_delegate) {
366 return connection_->SendStreamData(id, data, offset, fin, fec_protection,
367 ack_notifier_delegate);
368 }
369
WriteHeaders(QuicStreamId id,const SpdyHeaderBlock & headers,bool fin,QuicAckNotifier::DelegateInterface * ack_notifier_delegate)370 size_t QuicSession::WriteHeaders(
371 QuicStreamId id,
372 const SpdyHeaderBlock& headers,
373 bool fin,
374 QuicAckNotifier::DelegateInterface* ack_notifier_delegate) {
375 return headers_stream_->WriteHeaders(id, headers, fin, ack_notifier_delegate);
376 }
377
SendRstStream(QuicStreamId id,QuicRstStreamErrorCode error,QuicStreamOffset bytes_written)378 void QuicSession::SendRstStream(QuicStreamId id,
379 QuicRstStreamErrorCode error,
380 QuicStreamOffset bytes_written) {
381 if (connection()->connected()) {
382 // Only send a RST_STREAM frame if still connected.
383 connection_->SendRstStream(id, error, bytes_written);
384 }
385 CloseStreamInner(id, true);
386 }
387
SendGoAway(QuicErrorCode error_code,const string & reason)388 void QuicSession::SendGoAway(QuicErrorCode error_code, const string& reason) {
389 if (goaway_sent_) {
390 return;
391 }
392 goaway_sent_ = true;
393 connection_->SendGoAway(error_code, largest_peer_created_stream_id_, reason);
394 }
395
CloseStream(QuicStreamId stream_id)396 void QuicSession::CloseStream(QuicStreamId stream_id) {
397 CloseStreamInner(stream_id, false);
398 }
399
CloseStreamInner(QuicStreamId stream_id,bool locally_reset)400 void QuicSession::CloseStreamInner(QuicStreamId stream_id,
401 bool locally_reset) {
402 DVLOG(1) << ENDPOINT << "Closing stream " << stream_id;
403
404 DataStreamMap::iterator it = stream_map_.find(stream_id);
405 if (it == stream_map_.end()) {
406 DVLOG(1) << ENDPOINT << "Stream is already closed: " << stream_id;
407 return;
408 }
409 QuicDataStream* stream = it->second;
410
411 // Tell the stream that a RST has been sent.
412 if (locally_reset) {
413 stream->set_rst_sent(true);
414 }
415
416 closed_streams_.push_back(it->second);
417
418 // If we haven't received a FIN or RST for this stream, we need to keep track
419 // of the how many bytes the stream's flow controller believes it has
420 // received, for accurate connection level flow control accounting.
421 if (!stream->HasFinalReceivedByteOffset() &&
422 stream->flow_controller()->IsEnabled() &&
423 FLAGS_enable_quic_connection_flow_control_2) {
424 locally_closed_streams_highest_offset_[stream_id] =
425 stream->flow_controller()->highest_received_byte_offset();
426 }
427
428 stream_map_.erase(it);
429 stream->OnClose();
430 }
431
UpdateFlowControlOnFinalReceivedByteOffset(QuicStreamId stream_id,QuicStreamOffset final_byte_offset)432 void QuicSession::UpdateFlowControlOnFinalReceivedByteOffset(
433 QuicStreamId stream_id, QuicStreamOffset final_byte_offset) {
434 if (!FLAGS_enable_quic_connection_flow_control_2) {
435 return;
436 }
437
438 map<QuicStreamId, QuicStreamOffset>::iterator it =
439 locally_closed_streams_highest_offset_.find(stream_id);
440 if (it == locally_closed_streams_highest_offset_.end()) {
441 return;
442 }
443
444 DVLOG(1) << ENDPOINT << "Received final byte offset " << final_byte_offset
445 << " for stream " << stream_id;
446 uint64 offset_diff = final_byte_offset - it->second;
447 if (flow_controller_->UpdateHighestReceivedOffset(
448 flow_controller_->highest_received_byte_offset() + offset_diff)) {
449 // If the final offset violates flow control, close the connection now.
450 if (flow_controller_->FlowControlViolation()) {
451 connection_->SendConnectionClose(
452 QUIC_FLOW_CONTROL_RECEIVED_TOO_MUCH_DATA);
453 return;
454 }
455 }
456
457 flow_controller_->AddBytesConsumed(offset_diff);
458 locally_closed_streams_highest_offset_.erase(it);
459 }
460
IsEncryptionEstablished()461 bool QuicSession::IsEncryptionEstablished() {
462 return GetCryptoStream()->encryption_established();
463 }
464
IsCryptoHandshakeConfirmed()465 bool QuicSession::IsCryptoHandshakeConfirmed() {
466 return GetCryptoStream()->handshake_confirmed();
467 }
468
OnConfigNegotiated()469 void QuicSession::OnConfigNegotiated() {
470 connection_->SetFromConfig(config_);
471 QuicVersion version = connection()->version();
472 if (version < QUIC_VERSION_17) {
473 return;
474 }
475
476 if (version <= QUIC_VERSION_19) {
477 // QUIC_VERSION_17,18,19 don't support independent stream/session flow
478 // control windows.
479 if (config_.HasReceivedInitialFlowControlWindowBytes()) {
480 // Streams which were created before the SHLO was received (0-RTT
481 // requests) are now informed of the peer's initial flow control window.
482 uint32 new_window = config_.ReceivedInitialFlowControlWindowBytes();
483 OnNewStreamFlowControlWindow(new_window);
484 OnNewSessionFlowControlWindow(new_window);
485 }
486
487 return;
488 }
489
490 // QUIC_VERSION_20 and higher can have independent stream and session flow
491 // control windows.
492 if (config_.HasReceivedInitialStreamFlowControlWindowBytes()) {
493 // Streams which were created before the SHLO was received (0-RTT
494 // requests) are now informed of the peer's initial flow control window.
495 OnNewStreamFlowControlWindow(
496 config_.ReceivedInitialStreamFlowControlWindowBytes());
497 }
498 if (config_.HasReceivedInitialSessionFlowControlWindowBytes()) {
499 OnNewSessionFlowControlWindow(
500 config_.ReceivedInitialSessionFlowControlWindowBytes());
501 }
502 }
503
OnNewStreamFlowControlWindow(uint32 new_window)504 void QuicSession::OnNewStreamFlowControlWindow(uint32 new_window) {
505 if (new_window < kDefaultFlowControlSendWindow) {
506 LOG(ERROR)
507 << "Peer sent us an invalid stream flow control send window: "
508 << new_window << ", below default: " << kDefaultFlowControlSendWindow;
509 if (connection_->connected()) {
510 connection_->SendConnectionClose(QUIC_FLOW_CONTROL_INVALID_WINDOW);
511 }
512 return;
513 }
514
515 for (DataStreamMap::iterator it = stream_map_.begin();
516 it != stream_map_.end(); ++it) {
517 it->second->flow_controller()->UpdateSendWindowOffset(new_window);
518 }
519 }
520
OnNewSessionFlowControlWindow(uint32 new_window)521 void QuicSession::OnNewSessionFlowControlWindow(uint32 new_window) {
522 if (new_window < kDefaultFlowControlSendWindow) {
523 LOG(ERROR)
524 << "Peer sent us an invalid session flow control send window: "
525 << new_window << ", below default: " << kDefaultFlowControlSendWindow;
526 if (connection_->connected()) {
527 connection_->SendConnectionClose(QUIC_FLOW_CONTROL_INVALID_WINDOW);
528 }
529 return;
530 }
531
532 flow_controller_->UpdateSendWindowOffset(new_window);
533 }
534
OnCryptoHandshakeEvent(CryptoHandshakeEvent event)535 void QuicSession::OnCryptoHandshakeEvent(CryptoHandshakeEvent event) {
536 switch (event) {
537 // TODO(satyamshekhar): Move the logic of setting the encrypter/decrypter
538 // to QuicSession since it is the glue.
539 case ENCRYPTION_FIRST_ESTABLISHED:
540 break;
541
542 case ENCRYPTION_REESTABLISHED:
543 // Retransmit originally packets that were sent, since they can't be
544 // decrypted by the peer.
545 connection_->RetransmitUnackedPackets(INITIAL_ENCRYPTION_ONLY);
546 break;
547
548 case HANDSHAKE_CONFIRMED:
549 LOG_IF(DFATAL, !config_.negotiated()) << ENDPOINT
550 << "Handshake confirmed without parameter negotiation.";
551 // Discard originally encrypted packets, since they can't be decrypted by
552 // the peer.
553 connection_->NeuterUnencryptedPackets();
554 connection_->SetOverallConnectionTimeout(QuicTime::Delta::Infinite());
555 max_open_streams_ = config_.max_streams_per_connection();
556 break;
557
558 default:
559 LOG(ERROR) << ENDPOINT << "Got unknown handshake event: " << event;
560 }
561 }
562
OnCryptoHandshakeMessageSent(const CryptoHandshakeMessage & message)563 void QuicSession::OnCryptoHandshakeMessageSent(
564 const CryptoHandshakeMessage& message) {
565 }
566
OnCryptoHandshakeMessageReceived(const CryptoHandshakeMessage & message)567 void QuicSession::OnCryptoHandshakeMessageReceived(
568 const CryptoHandshakeMessage& message) {
569 }
570
config()571 QuicConfig* QuicSession::config() {
572 return &config_;
573 }
574
ActivateStream(QuicDataStream * stream)575 void QuicSession::ActivateStream(QuicDataStream* stream) {
576 DVLOG(1) << ENDPOINT << "num_streams: " << stream_map_.size()
577 << ". activating " << stream->id();
578 DCHECK_EQ(stream_map_.count(stream->id()), 0u);
579 stream_map_[stream->id()] = stream;
580 }
581
GetNextStreamId()582 QuicStreamId QuicSession::GetNextStreamId() {
583 QuicStreamId id = next_stream_id_;
584 next_stream_id_ += 2;
585 return id;
586 }
587
GetStream(const QuicStreamId stream_id)588 ReliableQuicStream* QuicSession::GetStream(const QuicStreamId stream_id) {
589 if (stream_id == kCryptoStreamId) {
590 return GetCryptoStream();
591 }
592 if (stream_id == kHeadersStreamId) {
593 return headers_stream_.get();
594 }
595 return GetDataStream(stream_id);
596 }
597
GetDataStream(const QuicStreamId stream_id)598 QuicDataStream* QuicSession::GetDataStream(const QuicStreamId stream_id) {
599 if (stream_id == kCryptoStreamId) {
600 DLOG(FATAL) << "Attempt to call GetDataStream with the crypto stream id";
601 return NULL;
602 }
603 if (stream_id == kHeadersStreamId) {
604 DLOG(FATAL) << "Attempt to call GetDataStream with the headers stream id";
605 return NULL;
606 }
607
608 DataStreamMap::iterator it = stream_map_.find(stream_id);
609 if (it != stream_map_.end()) {
610 return it->second;
611 }
612
613 if (IsClosedStream(stream_id)) {
614 return NULL;
615 }
616
617 if (stream_id % 2 == next_stream_id_ % 2) {
618 // We've received a frame for a locally-created stream that is not
619 // currently active. This is an error.
620 if (connection()->connected()) {
621 connection()->SendConnectionClose(QUIC_PACKET_FOR_NONEXISTENT_STREAM);
622 }
623 return NULL;
624 }
625
626 return GetIncomingDataStream(stream_id);
627 }
628
GetIncomingDataStream(QuicStreamId stream_id)629 QuicDataStream* QuicSession::GetIncomingDataStream(QuicStreamId stream_id) {
630 if (IsClosedStream(stream_id)) {
631 return NULL;
632 }
633
634 implicitly_created_streams_.erase(stream_id);
635 if (stream_id > largest_peer_created_stream_id_) {
636 if (stream_id - largest_peer_created_stream_id_ > kMaxStreamIdDelta) {
637 // We may already have sent a connection close due to multiple reset
638 // streams in the same packet.
639 if (connection()->connected()) {
640 LOG(ERROR) << "Trying to get stream: " << stream_id
641 << ", largest peer created stream: "
642 << largest_peer_created_stream_id_
643 << ", max delta: " << kMaxStreamIdDelta;
644 connection()->SendConnectionClose(QUIC_INVALID_STREAM_ID);
645 }
646 return NULL;
647 }
648 if (largest_peer_created_stream_id_ == 0) {
649 if (is_server()) {
650 largest_peer_created_stream_id_= 3;
651 } else {
652 largest_peer_created_stream_id_= 1;
653 }
654 }
655 for (QuicStreamId id = largest_peer_created_stream_id_ + 2;
656 id < stream_id;
657 id += 2) {
658 implicitly_created_streams_.insert(id);
659 }
660 largest_peer_created_stream_id_ = stream_id;
661 }
662 QuicDataStream* stream = CreateIncomingDataStream(stream_id);
663 if (stream == NULL) {
664 return NULL;
665 }
666 ActivateStream(stream);
667 return stream;
668 }
669
IsClosedStream(QuicStreamId id)670 bool QuicSession::IsClosedStream(QuicStreamId id) {
671 DCHECK_NE(0u, id);
672 if (id == kCryptoStreamId) {
673 return false;
674 }
675 if (id == kHeadersStreamId) {
676 return false;
677 }
678 if (ContainsKey(stream_map_, id)) {
679 // Stream is active
680 return false;
681 }
682 if (id % 2 == next_stream_id_ % 2) {
683 // Locally created streams are strictly in-order. If the id is in the
684 // range of created streams and it's not active, it must have been closed.
685 return id < next_stream_id_;
686 }
687 // For peer created streams, we also need to consider implicitly created
688 // streams.
689 return id <= largest_peer_created_stream_id_ &&
690 implicitly_created_streams_.count(id) == 0;
691 }
692
GetNumOpenStreams() const693 size_t QuicSession::GetNumOpenStreams() const {
694 return stream_map_.size() + implicitly_created_streams_.size();
695 }
696
MarkWriteBlocked(QuicStreamId id,QuicPriority priority)697 void QuicSession::MarkWriteBlocked(QuicStreamId id, QuicPriority priority) {
698 #ifndef NDEBUG
699 ReliableQuicStream* stream = GetStream(id);
700 if (stream != NULL) {
701 LOG_IF(DFATAL, priority != stream->EffectivePriority())
702 << ENDPOINT << "Stream " << id
703 << "Priorities do not match. Got: " << priority
704 << " Expected: " << stream->EffectivePriority();
705 } else {
706 LOG(DFATAL) << "Marking unknown stream " << id << " blocked.";
707 }
708 #endif
709
710 if (id == kCryptoStreamId) {
711 DCHECK(!has_pending_handshake_);
712 has_pending_handshake_ = true;
713 // TODO(jar): Be sure to use the highest priority for the crypto stream,
714 // perhaps by adding a "special" priority for it that is higher than
715 // kHighestPriority.
716 priority = kHighestPriority;
717 }
718 write_blocked_streams_.PushBack(id, priority);
719 }
720
HasDataToWrite() const721 bool QuicSession::HasDataToWrite() const {
722 return write_blocked_streams_.HasWriteBlockedCryptoOrHeadersStream() ||
723 write_blocked_streams_.HasWriteBlockedDataStreams() ||
724 connection_->HasQueuedData();
725 }
726
GetSSLInfo(SSLInfo * ssl_info) const727 bool QuicSession::GetSSLInfo(SSLInfo* ssl_info) const {
728 NOTIMPLEMENTED();
729 return false;
730 }
731
PostProcessAfterData()732 void QuicSession::PostProcessAfterData() {
733 STLDeleteElements(&closed_streams_);
734 closed_streams_.clear();
735 }
736
OnSuccessfulVersionNegotiation(const QuicVersion & version)737 void QuicSession::OnSuccessfulVersionNegotiation(const QuicVersion& version) {
738 if (version < QUIC_VERSION_19) {
739 flow_controller_->Disable();
740 }
741
742 // Inform all streams about the negotiated version. They may have been created
743 // with a different version.
744 for (DataStreamMap::iterator it = stream_map_.begin();
745 it != stream_map_.end(); ++it) {
746 if (version < QUIC_VERSION_17) {
747 it->second->flow_controller()->Disable();
748 }
749 }
750 }
751
752 } // namespace net
753