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