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