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