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_flow_controller.h"
  11 #include "net/ssl/ssl_info.h"
  12
  13 using base::StringPiece;
  14 using base::hash_map;
  15 using base::hash_set;
  16 using std::make_pair;
  17 using std::map;
  18 using std::max;
  19 using std::string;
  20 using std::vector;
  21
  22 namespace net {
  23
  24 #define ENDPOINT \
  25   (perspective() == Perspective::IS_SERVER ? "Server: " : " Client: ")
  26
  27 // We want to make sure we delete any closed streams in a safe manner.
  28 // To avoid deleting a stream in mid-operation, we have a simple shim between
  29 // us and the stream, so we can delete any streams when we return from
  30 // processing.
  31 //
  32 // We could just override the base methods, but this makes it easier to make
  33 // sure we don't miss any.
  34 class VisitorShim : public QuicConnectionVisitorInterface {
  35  public:
  36   explicit VisitorShim(QuicSession* session) : session_(session) {}
  37
  38   void OnStreamFrames(const vector<QuicStreamFrame>& frames) override {
  39     session_->OnStreamFrames(frames);
  40     session_->PostProcessAfterData();
  41   }
  42   void OnRstStream(const QuicRstStreamFrame& frame) override {
  43     session_->OnRstStream(frame);
  44     session_->PostProcessAfterData();
  45   }
  46
  47   void OnGoAway(const QuicGoAwayFrame& frame) override {
  48     session_->OnGoAway(frame);
  49     session_->PostProcessAfterData();
  50   }
  51
  52   void OnWindowUpdateFrames(
  53       const vector<QuicWindowUpdateFrame>& frames) override {
  54     session_->OnWindowUpdateFrames(frames);
  55     session_->PostProcessAfterData();
  56   }
  57
  58   void OnBlockedFrames(const vector<QuicBlockedFrame>& frames) override {
  59     session_->OnBlockedFrames(frames);
  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::OnStreamFrames(const vector<QuicStreamFrame>& frames) {
 138   for (size_t i = 0; i < frames.size() && connection_->connected(); ++i) {
 139     // TODO(rch) deal with the error case of stream id 0.
 140     const QuicStreamFrame& frame = frames[i];
 141     QuicStreamId stream_id = frame.stream_id;
 142     ReliableQuicStream* stream = GetStream(stream_id);
 143     if (!stream) {
 144       // The stream no longer exists, but we may still be interested in the
 145       // final stream byte offset sent by the peer. A frame with a FIN can give
 146       // us this offset.
 147       if (frame.fin) {
 148         QuicStreamOffset final_byte_offset = frame.offset + frame.data.size();
 149         UpdateFlowControlOnFinalReceivedByteOffset(stream_id,
 150                                                    final_byte_offset);
 151       }
 152
 153       continue;
 154     }
 155     stream->OnStreamFrame(frames[i]);
 156   }
 157 }
 158
 159 void QuicSession::OnRstStream(const QuicRstStreamFrame& frame) {
 160   if (ContainsKey(static_stream_map_, frame.stream_id)) {
 161     connection()->SendConnectionCloseWithDetails(
 162         QUIC_INVALID_STREAM_ID, "Attempt to reset a static stream");
 163     return;
 164   }
 165
 166   ReliableQuicStream* stream = GetDynamicStream(frame.stream_id);
 167   if (!stream) {
 168     // The RST frame contains the final byte offset for the stream: we can now
 169     // update the connection level flow controller if needed.
 170     UpdateFlowControlOnFinalReceivedByteOffset(frame.stream_id,
 171                                                frame.byte_offset);
 172     return;  // Errors are handled by GetStream.
 173   }
 174
 175   stream->OnStreamReset(frame);
 176 }
 177
 178 void QuicSession::OnGoAway(const QuicGoAwayFrame& frame) {
 179   DCHECK(frame.last_good_stream_id < next_stream_id_);
 180   goaway_received_ = true;
 181 }
 182
 183 void QuicSession::OnConnectionClosed(QuicErrorCode error, bool from_peer) {
 184   DCHECK(!connection_->connected());
 185   if (error_ == QUIC_NO_ERROR) {
 186     error_ = error;
 187   }
 188
 189   while (!dynamic_stream_map_.empty()) {
 190     StreamMap::iterator it = dynamic_stream_map_.begin();
 191     QuicStreamId id = it->first;
 192     it->second->OnConnectionClosed(error, from_peer);
 193     // The stream should call CloseStream as part of OnConnectionClosed.
 194     if (dynamic_stream_map_.find(id) != dynamic_stream_map_.end()) {
 195       LOG(DFATAL) << ENDPOINT
 196                   << "Stream failed to close under OnConnectionClosed";
 197       CloseStream(id);
 198     }
 199   }
 200 }
 201
 202 void QuicSession::OnSuccessfulVersionNegotiation(const QuicVersion& version) {
 203 }
 204
 205 void QuicSession::OnWindowUpdateFrames(
 206     const vector<QuicWindowUpdateFrame>& frames) {
 207   bool connection_window_updated = false;
 208   for (size_t i = 0; i < frames.size(); ++i) {
 209     // Stream may be closed by the time we receive a WINDOW_UPDATE, so we can't
 210     // assume that it still exists.
 211     QuicStreamId stream_id = frames[i].stream_id;
 212     if (stream_id == kConnectionLevelId) {
 213       // This is a window update that applies to the connection, rather than an
 214       // individual stream.
 215       DVLOG(1) << ENDPOINT
 216                << "Received connection level flow control window update with "
 217                   "byte offset: " << frames[i].byte_offset;
 218       if (flow_controller_.UpdateSendWindowOffset(frames[i].byte_offset)) {
 219         connection_window_updated = true;
 220       }
 221       continue;
 222     }
 223
 224     ReliableQuicStream* stream = GetStream(stream_id);
 225     if (stream) {
 226       stream->OnWindowUpdateFrame(frames[i]);
 227     }
 228   }
 229
 230   // Connection level flow control window has increased, so blocked streams can
 231   // write again.
 232   if (connection_window_updated) {
 233     OnCanWrite();
 234   }
 235 }
 236
 237 void QuicSession::OnBlockedFrames(const vector<QuicBlockedFrame>& frames) {
 238   for (size_t i = 0; i < frames.size(); ++i) {
 239     // TODO(rjshade): Compare our flow control receive windows for specified
 240     //                streams: if we have a large window then maybe something
 241     //                had gone wrong with the flow control accounting.
 242     DVLOG(1) << ENDPOINT << "Received BLOCKED frame with stream id: "
 243              << frames[i].stream_id;
 244   }
 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   goaway_sent_ = true;
 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     DVLOG(1) << ENDPOINT << "Stream is already closed: " << stream_id;
 358     return;
 359   }
 360   ReliableQuicStream* stream = it->second;
 361
 362   // Tell the stream that a RST has been sent.
 363   if (locally_reset) {
 364     stream->set_rst_sent(true);
 365   }
 366
 367   closed_streams_.push_back(it->second);
 368
 369   // If we haven't received a FIN or RST for this stream, we need to keep track
 370   // of the how many bytes the stream's flow controller believes it has
 371   // received, for accurate connection level flow control accounting.
 372   if (!stream->HasFinalReceivedByteOffset()) {
 373     locally_closed_streams_highest_offset_[stream_id] =
 374         stream->flow_controller()->highest_received_byte_offset();
 375   }
 376
 377   dynamic_stream_map_.erase(it);
 378   stream->OnClose();
 379   // Decrease the number of streams being emulated when a new one is opened.
 380   connection_->SetNumOpenStreams(dynamic_stream_map_.size());
 381 }
 382
 383 void QuicSession::UpdateFlowControlOnFinalReceivedByteOffset(
 384     QuicStreamId stream_id, QuicStreamOffset final_byte_offset) {
 385   map<QuicStreamId, QuicStreamOffset>::iterator it =
 386       locally_closed_streams_highest_offset_.find(stream_id);
 387   if (it == locally_closed_streams_highest_offset_.end()) {
 388     return;
 389   }
 390
 391   DVLOG(1) << ENDPOINT << "Received final byte offset " << final_byte_offset
 392            << " for stream " << stream_id;
 393   QuicByteCount offset_diff = final_byte_offset - it->second;
 394   if (flow_controller_.UpdateHighestReceivedOffset(
 395           flow_controller_.highest_received_byte_offset() + offset_diff)) {
 396     // If the final offset violates flow control, close the connection now.
 397     if (flow_controller_.FlowControlViolation()) {
 398       connection_->SendConnectionClose(
 399           QUIC_FLOW_CONTROL_RECEIVED_TOO_MUCH_DATA);
 400       return;
 401     }
 402   }
 403
 404   flow_controller_.AddBytesConsumed(offset_diff);
 405   locally_closed_streams_highest_offset_.erase(it);
 406 }
 407
 408 bool QuicSession::IsEncryptionEstablished() {
 409   return GetCryptoStream()->encryption_established();
 410 }
 411
 412 bool QuicSession::IsCryptoHandshakeConfirmed() {
 413   return GetCryptoStream()->handshake_confirmed();
 414 }
 415
 416 void QuicSession::OnConfigNegotiated() {
 417   connection_->SetFromConfig(config_);
 418
 419   uint32 max_streams = config_.MaxStreamsPerConnection();
 420   if (perspective() == Perspective::IS_SERVER) {
 421     // A server should accept a small number of additional streams beyond the
 422     // limit sent to the client. This helps avoid early connection termination
 423     // when FIN/RSTs for old streams are lost or arrive out of order.
 424     // Use a minimum number of additional streams, or a percentage increase,
 425     // whichever is larger.
 426     max_streams =
 427         max(max_streams + kMaxStreamsMinimumIncrement,
 428             static_cast<uint32>(max_streams * kMaxStreamsMultiplier));
 429
 430     if (config_.HasReceivedConnectionOptions() &&
 431         ContainsQuicTag(config_.ReceivedConnectionOptions(), kAFCW)) {
 432       EnableAutoTuneReceiveWindow();
 433     }
 434   }
 435   set_max_open_streams(max_streams);
 436
 437   if (config_.HasReceivedInitialStreamFlowControlWindowBytes()) {
 438     // Streams which were created before the SHLO was received (0-RTT
 439     // requests) are now informed of the peer's initial flow control window.
 440     OnNewStreamFlowControlWindow(
 441         config_.ReceivedInitialStreamFlowControlWindowBytes());
 442   }
 443   if (config_.HasReceivedInitialSessionFlowControlWindowBytes()) {
 444     OnNewSessionFlowControlWindow(
 445         config_.ReceivedInitialSessionFlowControlWindowBytes());
 446   }
 447 }
 448
 449 void QuicSession::EnableAutoTuneReceiveWindow() {
 450   flow_controller_.set_auto_tune_receive_window(true);
 451   // Inform all existing streams about the new window.
 452   for (auto const& kv : static_stream_map_) {
 453     kv.second->flow_controller()->set_auto_tune_receive_window(true);
 454   }
 455   for (auto const& kv : dynamic_stream_map_) {
 456     kv.second->flow_controller()->set_auto_tune_receive_window(true);
 457   }
 458 }
 459
 460 void QuicSession::OnNewStreamFlowControlWindow(QuicStreamOffset new_window) {
 461   if (new_window < kMinimumFlowControlSendWindow) {
 462     LOG(ERROR) << "Peer sent us an invalid stream flow control send window: "
 463                << new_window
 464                << ", below default: " << kMinimumFlowControlSendWindow;
 465     if (connection_->connected()) {
 466       connection_->SendConnectionClose(QUIC_FLOW_CONTROL_INVALID_WINDOW);
 467     }
 468     return;
 469   }
 470
 471   // Inform all existing streams about the new window.
 472   for (auto const& kv : static_stream_map_) {
 473     kv.second->UpdateSendWindowOffset(new_window);
 474   }
 475   for (auto const& kv : dynamic_stream_map_) {
 476     kv.second->UpdateSendWindowOffset(new_window);
 477   }
 478 }
 479
 480 void QuicSession::OnNewSessionFlowControlWindow(QuicStreamOffset new_window) {
 481   if (new_window < kMinimumFlowControlSendWindow) {
 482     LOG(ERROR) << "Peer sent us an invalid session flow control send window: "
 483                << new_window
 484                << ", below default: " << kMinimumFlowControlSendWindow;
 485     if (connection_->connected()) {
 486       connection_->SendConnectionClose(QUIC_FLOW_CONTROL_INVALID_WINDOW);
 487     }
 488     return;
 489   }
 490
 491   flow_controller_.UpdateSendWindowOffset(new_window);
 492 }
 493
 494 void QuicSession::OnCryptoHandshakeEvent(CryptoHandshakeEvent event) {
 495   switch (event) {
 496     // TODO(satyamshekhar): Move the logic of setting the encrypter/decrypter
 497     // to QuicSession since it is the glue.
 498     case ENCRYPTION_FIRST_ESTABLISHED:
 499       break;
 500
 501     case ENCRYPTION_REESTABLISHED:
 502       // Retransmit originally packets that were sent, since they can't be
 503       // decrypted by the peer.
 504       connection_->RetransmitUnackedPackets(ALL_INITIAL_RETRANSMISSION);
 505       break;
 506
 507     case HANDSHAKE_CONFIRMED:
 508       LOG_IF(DFATAL, !config_.negotiated()) << ENDPOINT
 509           << "Handshake confirmed without parameter negotiation.";
 510       // Discard originally encrypted packets, since they can't be decrypted by
 511       // the peer.
 512       connection_->NeuterUnencryptedPackets();
 513       break;
 514
 515     default:
 516       LOG(ERROR) << ENDPOINT << "Got unknown handshake event: " << event;
 517   }
 518 }
 519
 520 void QuicSession::OnCryptoHandshakeMessageSent(
 521     const CryptoHandshakeMessage& message) {
 522 }
 523
 524 void QuicSession::OnCryptoHandshakeMessageReceived(
 525     const CryptoHandshakeMessage& message) {
 526 }
 527
 528 QuicConfig* QuicSession::config() {
 529   return &config_;
 530 }
 531
 532 void QuicSession::ActivateStream(ReliableQuicStream* stream) {
 533   DVLOG(1) << ENDPOINT << "num_streams: " << dynamic_stream_map_.size()
 534            << ". activating " << stream->id();
 535   DCHECK_EQ(dynamic_stream_map_.count(stream->id()), 0u);
 536   dynamic_stream_map_[stream->id()] = stream;
 537   // Increase the number of streams being emulated when a new one is opened.
 538   connection_->SetNumOpenStreams(dynamic_stream_map_.size());
 539 }
 540
 541 QuicStreamId QuicSession::GetNextStreamId() {
 542   QuicStreamId id = next_stream_id_;
 543   next_stream_id_ += 2;
 544   return id;
 545 }
 546
 547 ReliableQuicStream* QuicSession::GetStream(const QuicStreamId stream_id) {
 548   StreamMap::iterator it = static_stream_map_.find(stream_id);
 549   if (it != static_stream_map_.end()) {
 550     return it->second;
 551   }
 552   return GetDynamicStream(stream_id);
 553 }
 554
 555 ReliableQuicStream* QuicSession::GetDynamicStream(
 556     const QuicStreamId stream_id) {
 557   if (static_stream_map_.find(stream_id) != static_stream_map_.end()) {
 558     DLOG(FATAL) << "Attempt to call GetDynamicStream for a static stream";
 559     return nullptr;
 560   }
 561
 562   StreamMap::iterator it = dynamic_stream_map_.find(stream_id);
 563   if (it != dynamic_stream_map_.end()) {
 564     return it->second;
 565   }
 566
 567   if (IsClosedStream(stream_id)) {
 568     return nullptr;
 569   }
 570
 571   if (stream_id % 2 == next_stream_id_ % 2) {
 572     // We've received a frame for a locally-created stream that is not
 573     // currently active.  This is an error.
 574     if (connection()->connected()) {
 575       connection()->SendConnectionClose(QUIC_PACKET_FOR_NONEXISTENT_STREAM);
 576     }
 577     return nullptr;
 578   }
 579
 580   return GetIncomingDynamicStream(stream_id);
 581 }
 582
 583 ReliableQuicStream* QuicSession::GetIncomingDynamicStream(
 584     QuicStreamId stream_id) {
 585   if (IsClosedStream(stream_id)) {
 586     return nullptr;
 587   }
 588
 589   implicitly_created_streams_.erase(stream_id);
 590   if (stream_id > largest_peer_created_stream_id_) {
 591     if (stream_id - largest_peer_created_stream_id_ > kMaxStreamIdDelta) {
 592       // We may already have sent a connection close due to multiple reset
 593       // streams in the same packet.
 594       if (connection()->connected()) {
 595         LOG(ERROR) << "Trying to get stream: " << stream_id
 596                    << ", largest peer created stream: "
 597                    << largest_peer_created_stream_id_
 598                    << ", max delta: " << kMaxStreamIdDelta;
 599         connection()->SendConnectionClose(QUIC_INVALID_STREAM_ID);
 600       }
 601       return nullptr;
 602     }
 603     for (QuicStreamId id = largest_peer_created_stream_id_ + 2;
 604          id < stream_id;
 605          id += 2) {
 606       implicitly_created_streams_.insert(id);
 607     }
 608     largest_peer_created_stream_id_ = stream_id;
 609   }
 610   ReliableQuicStream* stream = CreateIncomingDynamicStream(stream_id);
 611   if (stream == nullptr) {
 612     return nullptr;
 613   }
 614   ActivateStream(stream);
 615   return stream;
 616 }
 617
 618 void QuicSession::set_max_open_streams(size_t max_open_streams) {
 619   DVLOG(1) << "Setting max_open_streams_ to " << max_open_streams;
 620   max_open_streams_ = max_open_streams;
 621 }
 622
 623 bool QuicSession::IsClosedStream(QuicStreamId id) {
 624   DCHECK_NE(0u, id);
 625   if (ContainsKey(static_stream_map_, id) ||
 626       ContainsKey(dynamic_stream_map_, id)) {
 627     // Stream is active
 628     return false;
 629   }
 630   if (id % 2 == next_stream_id_ % 2) {
 631     // Locally created streams are strictly in-order.  If the id is in the
 632     // range of created streams and it's not active, it must have been closed.
 633     return id < next_stream_id_;
 634   }
 635   // For peer created streams, we also need to consider implicitly created
 636   // streams.
 637   return id <= largest_peer_created_stream_id_ &&
 638       !ContainsKey(implicitly_created_streams_, id);
 639 }
 640
 641 size_t QuicSession::GetNumOpenStreams() const {
 642   return dynamic_stream_map_.size() + implicitly_created_streams_.size();
 643 }
 644
 645 void QuicSession::MarkWriteBlocked(QuicStreamId id, QuicPriority priority) {
 646 #ifndef NDEBUG
 647   ReliableQuicStream* stream = GetStream(id);
 648   if (stream != nullptr) {
 649     LOG_IF(DFATAL, priority != stream->EffectivePriority())
 650         << ENDPOINT << "Stream " << id
 651         << "Priorities do not match.  Got: " << priority
 652         << " Expected: " << stream->EffectivePriority();
 653   } else {
 654     LOG(DFATAL) << "Marking unknown stream " << id << " blocked.";
 655   }
 656 #endif
 657
 658   if (id == kCryptoStreamId) {
 659     DCHECK(!has_pending_handshake_);
 660     has_pending_handshake_ = true;
 661     // TODO(jar): Be sure to use the highest priority for the crypto stream,
 662     // perhaps by adding a "special" priority for it that is higher than
 663     // kHighestPriority.
 664     priority = kHighestPriority;
 665   }
 666   write_blocked_streams_.PushBack(id, priority);
 667 }
 668
 669 bool QuicSession::HasDataToWrite() const {
 670   return write_blocked_streams_.HasWriteBlockedCryptoOrHeadersStream() ||
 671          write_blocked_streams_.HasWriteBlockedDataStreams() ||
 672          connection_->HasQueuedData();
 673 }
 674
 675 void QuicSession::PostProcessAfterData() {
 676   STLDeleteElements(&closed_streams_);
 677
 678   if (connection()->connected() &&
 679       locally_closed_streams_highest_offset_.size() > max_open_streams_) {
 680     // A buggy client may fail to send FIN/RSTs. Don't tolerate this.
 681     connection_->SendConnectionClose(QUIC_TOO_MANY_UNFINISHED_STREAMS);
 682   }
 683 }
 684
 685 bool QuicSession::IsConnectionFlowControlBlocked() const {
 686   return flow_controller_.IsBlocked();
 687 }
 688
 689 bool QuicSession::IsStreamFlowControlBlocked() {
 690   for (auto const& kv : static_stream_map_) {
 691     if (kv.second->flow_controller()->IsBlocked()) {
 692       return true;
 693     }
 694   }
 695   for (auto const& kv : dynamic_stream_map_) {
 696     if (kv.second->flow_controller()->IsBlocked()) {
 697       return true;
 698     }
 699   }
 700   return false;
 701 }
 702
 703 }  // namespace net