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