net/quic/quic_session.cc

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