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