net/quic/quic_session.cc

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