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