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