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       goaway_received_(false),
 108       goaway_sent_(false),
 109       has_pending_handshake_(false) {
 110   if (connection_->version() == QUIC_VERSION_19) {
 111     flow_controller_.reset(new QuicFlowController(
 112         connection_.get(), 0, is_server(), kDefaultFlowControlSendWindow,
 113         config_.GetInitialFlowControlWindowToSend(),
 114         config_.GetInitialFlowControlWindowToSend()));
 115   } else {
 116     flow_controller_.reset(new QuicFlowController(
 117         connection_.get(), 0, is_server(), kDefaultFlowControlSendWindow,
 118         config_.GetInitialSessionFlowControlWindowToSend(),
 119         config_.GetInitialSessionFlowControlWindowToSend()));
 120   }
 121 }
 122
 123 void QuicSession::InitializeSession() {
 124   connection_->set_visitor(visitor_shim_.get());
 125   connection_->SetFromConfig(config_);
 126   headers_stream_.reset(new QuicHeadersStream(this));
 127 }
 128
 129 QuicSession::~QuicSession() {
 130   STLDeleteElements(&closed_streams_);
 131   STLDeleteValues(&stream_map_);
 132
 133   DLOG_IF(WARNING,
 134           locally_closed_streams_highest_offset_.size() > max_open_streams_)
 135       << "Surprisingly high number of locally closed streams still waiting for "
 136          "final byte offset: " << locally_closed_streams_highest_offset_.size();
 137 }
 138
 139 void QuicSession::OnStreamFrames(const vector<QuicStreamFrame>& frames) {
 140   for (size_t i = 0; i < frames.size(); ++i) {
 141     // TODO(rch) deal with the error case of stream id 0.
 142     const QuicStreamFrame& frame = frames[i];
 143     QuicStreamId stream_id = frame.stream_id;
 144     ReliableQuicStream* stream = GetStream(stream_id);
 145     if (!stream) {
 146       // The stream no longer exists, but we may still be interested in the
 147       // final stream byte offset sent by the peer. A frame with a FIN can give
 148       // us this offset.
 149       if (frame.fin) {
 150         QuicStreamOffset final_byte_offset =
 151             frame.offset + frame.data.TotalBufferSize();
 152         UpdateFlowControlOnFinalReceivedByteOffset(stream_id,
 153                                                    final_byte_offset);
 154       }
 155
 156       continue;
 157     }
 158     stream->OnStreamFrame(frames[i]);
 159   }
 160 }
 161
 162 void QuicSession::OnStreamHeaders(QuicStreamId stream_id,
 163                                   StringPiece headers_data) {
 164   QuicDataStream* stream = GetDataStream(stream_id);
 165   if (!stream) {
 166     // It's quite possible to receive headers after a stream has been reset.
 167     return;
 168   }
 169   stream->OnStreamHeaders(headers_data);
 170 }
 171
 172 void QuicSession::OnStreamHeadersPriority(QuicStreamId stream_id,
 173                                           QuicPriority priority) {
 174   QuicDataStream* stream = GetDataStream(stream_id);
 175   if (!stream) {
 176     // It's quite possible to receive headers after a stream has been reset.
 177     return;
 178   }
 179   stream->OnStreamHeadersPriority(priority);
 180 }
 181
 182 void QuicSession::OnStreamHeadersComplete(QuicStreamId stream_id,
 183                                           bool fin,
 184                                           size_t frame_len) {
 185   QuicDataStream* stream = GetDataStream(stream_id);
 186   if (!stream) {
 187     // It's quite possible to receive headers after a stream has been reset.
 188     return;
 189   }
 190   stream->OnStreamHeadersComplete(fin, frame_len);
 191 }
 192
 193 void QuicSession::OnRstStream(const QuicRstStreamFrame& frame) {
 194   if (frame.stream_id == kCryptoStreamId) {
 195     connection()->SendConnectionCloseWithDetails(
 196         QUIC_INVALID_STREAM_ID,
 197         "Attempt to reset the crypto stream");
 198     return;
 199   }
 200   if (frame.stream_id == kHeadersStreamId) {
 201     connection()->SendConnectionCloseWithDetails(
 202         QUIC_INVALID_STREAM_ID,
 203         "Attempt to reset the headers stream");
 204     return;
 205   }
 206
 207   QuicDataStream* stream = GetDataStream(frame.stream_id);
 208   if (!stream) {
 209     // The RST frame contains the final byte offset for the stream: we can now
 210     // update the connection level flow controller if needed.
 211     UpdateFlowControlOnFinalReceivedByteOffset(frame.stream_id,
 212                                                frame.byte_offset);
 213     return;  // Errors are handled by GetStream.
 214   }
 215
 216   stream->OnStreamReset(frame);
 217 }
 218
 219 void QuicSession::OnGoAway(const QuicGoAwayFrame& frame) {
 220   DCHECK(frame.last_good_stream_id < next_stream_id_);
 221   goaway_received_ = true;
 222 }
 223
 224 void QuicSession::OnConnectionClosed(QuicErrorCode error, bool from_peer) {
 225   DCHECK(!connection_->connected());
 226   if (error_ == QUIC_NO_ERROR) {
 227     error_ = error;
 228   }
 229
 230   while (!stream_map_.empty()) {
 231     DataStreamMap::iterator it = stream_map_.begin();
 232     QuicStreamId id = it->first;
 233     it->second->OnConnectionClosed(error, from_peer);
 234     // The stream should call CloseStream as part of OnConnectionClosed.
 235     if (stream_map_.find(id) != stream_map_.end()) {
 236       LOG(DFATAL) << ENDPOINT
 237                   << "Stream failed to close under OnConnectionClosed";
 238       CloseStream(id);
 239     }
 240   }
 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     if (connection_->version() < QUIC_VERSION_21 &&
 263         (stream_id == kCryptoStreamId || stream_id == kHeadersStreamId)) {
 264       DLOG(DFATAL) << "WindowUpdate for stream " << stream_id << " in version "
 265                    << QuicVersionToString(connection_->version());
 266       return;
 267     }
 268
 269     ReliableQuicStream* stream = GetStream(stream_id);
 270     if (stream) {
 271       stream->OnWindowUpdateFrame(frames[i]);
 272     }
 273   }
 274
 275   // Connection level flow control window has increased, so blocked streams can
 276   // write again.
 277   if (connection_window_updated) {
 278     OnCanWrite();
 279   }
 280 }
 281
 282 void QuicSession::OnBlockedFrames(const vector<QuicBlockedFrame>& frames) {
 283   for (size_t i = 0; i < frames.size(); ++i) {
 284     // TODO(rjshade): Compare our flow control receive windows for specified
 285     //                streams: if we have a large window then maybe something
 286     //                had gone wrong with the flow control accounting.
 287     DVLOG(1) << ENDPOINT << "Received BLOCKED frame with stream id: "
 288              << frames[i].stream_id;
 289   }
 290 }
 291
 292 void QuicSession::OnCanWrite() {
 293   // We limit the number of writes to the number of pending streams. If more
 294   // streams become pending, WillingAndAbleToWrite will be true, which will
 295   // cause the connection to request resumption before yielding to other
 296   // connections.
 297   size_t num_writes = write_blocked_streams_.NumBlockedStreams();
 298   if (flow_controller_->IsBlocked()) {
 299     // If we are connection level flow control blocked, then only allow the
 300     // crypto and headers streams to try writing as all other streams will be
 301     // blocked.
 302     num_writes = 0;
 303     if (write_blocked_streams_.crypto_stream_blocked()) {
 304       num_writes += 1;
 305     }
 306     if (write_blocked_streams_.headers_stream_blocked()) {
 307       num_writes += 1;
 308     }
 309   }
 310   if (num_writes == 0) {
 311     return;
 312   }
 313
 314   QuicConnection::ScopedPacketBundler ack_bundler(
 315       connection_.get(), QuicConnection::NO_ACK);
 316   for (size_t i = 0; i < num_writes; ++i) {
 317     if (!(write_blocked_streams_.HasWriteBlockedCryptoOrHeadersStream() ||
 318           write_blocked_streams_.HasWriteBlockedDataStreams())) {
 319       // Writing one stream removed another!? Something's broken.
 320       LOG(DFATAL) << "WriteBlockedStream is missing";
 321       connection_->CloseConnection(QUIC_INTERNAL_ERROR, false);
 322       return;
 323     }
 324     if (!connection_->CanWriteStreamData()) {
 325       return;
 326     }
 327     QuicStreamId stream_id = write_blocked_streams_.PopFront();
 328     if (stream_id == kCryptoStreamId) {
 329       has_pending_handshake_ = false;  // We just popped it.
 330     }
 331     ReliableQuicStream* stream = GetStream(stream_id);
 332     if (stream != nullptr && !stream->flow_controller()->IsBlocked()) {
 333       // If the stream can't write all bytes, it'll re-add itself to the blocked
 334       // list.
 335       stream->OnCanWrite();
 336     }
 337   }
 338 }
 339
 340 bool QuicSession::WillingAndAbleToWrite() const {
 341   // If the crypto or headers streams are blocked, we want to schedule a write -
 342   // they don't get blocked by connection level flow control. Otherwise only
 343   // schedule a write if we are not flow control blocked at the connection
 344   // level.
 345   return write_blocked_streams_.HasWriteBlockedCryptoOrHeadersStream() ||
 346          (!flow_controller_->IsBlocked() &&
 347           write_blocked_streams_.HasWriteBlockedDataStreams());
 348 }
 349
 350 bool QuicSession::HasPendingHandshake() const {
 351   return has_pending_handshake_;
 352 }
 353
 354 bool QuicSession::HasOpenDataStreams() const {
 355   return GetNumOpenStreams() > 0;
 356 }
 357
 358 QuicConsumedData QuicSession::WritevData(
 359     QuicStreamId id,
 360     const IOVector& data,
 361     QuicStreamOffset offset,
 362     bool fin,
 363     FecProtection fec_protection,
 364     QuicAckNotifier::DelegateInterface* ack_notifier_delegate) {
 365   return connection_->SendStreamData(id, data, offset, fin, fec_protection,
 366                                      ack_notifier_delegate);
 367 }
 368
 369 size_t QuicSession::WriteHeaders(
 370     QuicStreamId id,
 371     const SpdyHeaderBlock& headers,
 372     bool fin,
 373     QuicAckNotifier::DelegateInterface* ack_notifier_delegate) {
 374   return headers_stream_->WriteHeaders(id, headers, fin, ack_notifier_delegate);
 375 }
 376
 377 void QuicSession::SendRstStream(QuicStreamId id,
 378                                 QuicRstStreamErrorCode error,
 379                                 QuicStreamOffset bytes_written) {
 380   if (connection()->connected()) {
 381     // Only send a RST_STREAM frame if still connected.
 382     connection_->SendRstStream(id, error, bytes_written);
 383   }
 384   CloseStreamInner(id, true);
 385 }
 386
 387 void QuicSession::SendGoAway(QuicErrorCode error_code, const string& reason) {
 388   if (goaway_sent_) {
 389     return;
 390   }
 391   goaway_sent_ = true;
 392   connection_->SendGoAway(error_code, largest_peer_created_stream_id_, reason);
 393 }
 394
 395 void QuicSession::CloseStream(QuicStreamId stream_id) {
 396   CloseStreamInner(stream_id, false);
 397 }
 398
 399 void QuicSession::CloseStreamInner(QuicStreamId stream_id,
 400                                    bool locally_reset) {
 401   DVLOG(1) << ENDPOINT << "Closing stream " << stream_id;
 402
 403   DataStreamMap::iterator it = stream_map_.find(stream_id);
 404   if (it == stream_map_.end()) {
 405     DVLOG(1) << ENDPOINT << "Stream is already closed: " << stream_id;
 406     return;
 407   }
 408   QuicDataStream* stream = it->second;
 409
 410   // Tell the stream that a RST has been sent.
 411   if (locally_reset) {
 412     stream->set_rst_sent(true);
 413   }
 414
 415   closed_streams_.push_back(it->second);
 416
 417   // If we haven't received a FIN or RST for this stream, we need to keep track
 418   // of the how many bytes the stream's flow controller believes it has
 419   // received, for accurate connection level flow control accounting.
 420   if (!stream->HasFinalReceivedByteOffset() &&
 421       stream->flow_controller()->IsEnabled()) {
 422     locally_closed_streams_highest_offset_[stream_id] =
 423         stream->flow_controller()->highest_received_byte_offset();
 424   }
 425
 426   stream_map_.erase(it);
 427   stream->OnClose();
 428   // Decrease the number of streams being emulated when a new one is opened.
 429   connection_->SetNumOpenStreams(stream_map_.size());
 430 }
 431
 432 void QuicSession::UpdateFlowControlOnFinalReceivedByteOffset(
 433     QuicStreamId stream_id, QuicStreamOffset final_byte_offset) {
 434   map<QuicStreamId, QuicStreamOffset>::iterator it =
 435       locally_closed_streams_highest_offset_.find(stream_id);
 436   if (it == locally_closed_streams_highest_offset_.end()) {
 437     return;
 438   }
 439
 440   DVLOG(1) << ENDPOINT << "Received final byte offset " << final_byte_offset
 441            << " for stream " << stream_id;
 442   QuicByteCount offset_diff = final_byte_offset - it->second;
 443   if (flow_controller_->UpdateHighestReceivedOffset(
 444       flow_controller_->highest_received_byte_offset() + offset_diff)) {
 445     // If the final offset violates flow control, close the connection now.
 446     if (flow_controller_->FlowControlViolation()) {
 447       connection_->SendConnectionClose(
 448           QUIC_FLOW_CONTROL_RECEIVED_TOO_MUCH_DATA);
 449       return;
 450     }
 451   }
 452
 453   flow_controller_->AddBytesConsumed(offset_diff);
 454   locally_closed_streams_highest_offset_.erase(it);
 455 }
 456
 457 bool QuicSession::IsEncryptionEstablished() {
 458   return GetCryptoStream()->encryption_established();
 459 }
 460
 461 bool QuicSession::IsCryptoHandshakeConfirmed() {
 462   return GetCryptoStream()->handshake_confirmed();
 463 }
 464
 465 void QuicSession::OnConfigNegotiated() {
 466   connection_->SetFromConfig(config_);
 467   QuicVersion version = connection()->version();
 468
 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   if (version == QUIC_VERSION_19) {
 483     // QUIC_VERSION_19 doesn't support independent stream/session flow
 484     // control windows.
 485     if (config_.HasReceivedInitialFlowControlWindowBytes()) {
 486       // Streams which were created before the SHLO was received (0-RTT
 487       // requests) are now informed of the peer's initial flow control window.
 488       QuicStreamOffset new_window =
 489           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(QuicStreamOffset 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(QuicStreamOffset 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       break;
 567
 568     default:
 569       LOG(ERROR) << ENDPOINT << "Got unknown handshake event: " << event;
 570   }
 571 }
 572
 573 void QuicSession::OnCryptoHandshakeMessageSent(
 574     const CryptoHandshakeMessage& message) {
 575 }
 576
 577 void QuicSession::OnCryptoHandshakeMessageReceived(
 578     const CryptoHandshakeMessage& message) {
 579 }
 580
 581 QuicConfig* QuicSession::config() {
 582   return &config_;
 583 }
 584
 585 void QuicSession::ActivateStream(QuicDataStream* stream) {
 586   DVLOG(1) << ENDPOINT << "num_streams: " << stream_map_.size()
 587            << ". activating " << stream->id();
 588   DCHECK_EQ(stream_map_.count(stream->id()), 0u);
 589   stream_map_[stream->id()] = stream;
 590   // Increase the number of streams being emulated when a new one is opened.
 591   connection_->SetNumOpenStreams(stream_map_.size());
 592 }
 593
 594 QuicStreamId QuicSession::GetNextStreamId() {
 595   QuicStreamId id = next_stream_id_;
 596   next_stream_id_ += 2;
 597   return id;
 598 }
 599
 600 ReliableQuicStream* QuicSession::GetStream(const QuicStreamId stream_id) {
 601   if (stream_id == kCryptoStreamId) {
 602     return GetCryptoStream();
 603   }
 604   if (stream_id == kHeadersStreamId) {
 605     return headers_stream_.get();
 606   }
 607   return GetDataStream(stream_id);
 608 }
 609
 610 QuicDataStream* QuicSession::GetDataStream(const QuicStreamId stream_id) {
 611   if (stream_id == kCryptoStreamId) {
 612     DLOG(FATAL) << "Attempt to call GetDataStream with the crypto stream id";
 613     return nullptr;
 614   }
 615   if (stream_id == kHeadersStreamId) {
 616     DLOG(FATAL) << "Attempt to call GetDataStream with the headers stream id";
 617     return nullptr;
 618   }
 619
 620   DataStreamMap::iterator it = stream_map_.find(stream_id);
 621   if (it != stream_map_.end()) {
 622     return it->second;
 623   }
 624
 625   if (IsClosedStream(stream_id)) {
 626     return nullptr;
 627   }
 628
 629   if (stream_id % 2 == next_stream_id_ % 2) {
 630     // We've received a frame for a locally-created stream that is not
 631     // currently active.  This is an error.
 632     if (connection()->connected()) {
 633       connection()->SendConnectionClose(QUIC_PACKET_FOR_NONEXISTENT_STREAM);
 634     }
 635     return nullptr;
 636   }
 637
 638   return GetIncomingDataStream(stream_id);
 639 }
 640
 641 QuicDataStream* QuicSession::GetIncomingDataStream(QuicStreamId stream_id) {
 642   if (IsClosedStream(stream_id)) {
 643     return nullptr;
 644   }
 645
 646   implicitly_created_streams_.erase(stream_id);
 647   if (stream_id > largest_peer_created_stream_id_) {
 648     if (stream_id - largest_peer_created_stream_id_ > kMaxStreamIdDelta) {
 649       // We may already have sent a connection close due to multiple reset
 650       // streams in the same packet.
 651       if (connection()->connected()) {
 652         LOG(ERROR) << "Trying to get stream: " << stream_id
 653                    << ", largest peer created stream: "
 654                    << largest_peer_created_stream_id_
 655                    << ", max delta: " << kMaxStreamIdDelta;
 656         connection()->SendConnectionClose(QUIC_INVALID_STREAM_ID);
 657       }
 658       return nullptr;
 659     }
 660     if (largest_peer_created_stream_id_ == 0) {
 661       if (is_server()) {
 662         largest_peer_created_stream_id_ = 3;
 663       } else {
 664         largest_peer_created_stream_id_ = 1;
 665       }
 666     }
 667     for (QuicStreamId id = largest_peer_created_stream_id_ + 2;
 668          id < stream_id;
 669          id += 2) {
 670       implicitly_created_streams_.insert(id);
 671     }
 672     largest_peer_created_stream_id_ = stream_id;
 673   }
 674   QuicDataStream* stream = CreateIncomingDataStream(stream_id);
 675   if (stream == nullptr) {
 676     return nullptr;
 677   }
 678   ActivateStream(stream);
 679   return stream;
 680 }
 681
 682 void QuicSession::set_max_open_streams(size_t max_open_streams) {
 683   DVLOG(1) << "Setting max_open_streams_ to " << max_open_streams;
 684   max_open_streams_ = max_open_streams;
 685 }
 686
 687 bool QuicSession::IsClosedStream(QuicStreamId id) {
 688   DCHECK_NE(0u, id);
 689   if (id == kCryptoStreamId) {
 690     return false;
 691   }
 692   if (id == kHeadersStreamId) {
 693     return false;
 694   }
 695   if (ContainsKey(stream_map_, id)) {
 696     // Stream is active
 697     return false;
 698   }
 699   if (id % 2 == next_stream_id_ % 2) {
 700     // Locally created streams are strictly in-order.  If the id is in the
 701     // range of created streams and it's not active, it must have been closed.
 702     return id < next_stream_id_;
 703   }
 704   // For peer created streams, we also need to consider implicitly created
 705   // streams.
 706   return id <= largest_peer_created_stream_id_ &&
 707       !ContainsKey(implicitly_created_streams_, id);
 708 }
 709
 710 size_t QuicSession::GetNumOpenStreams() const {
 711   return stream_map_.size() + implicitly_created_streams_.size();
 712 }
 713
 714 void QuicSession::MarkWriteBlocked(QuicStreamId id, QuicPriority priority) {
 715 #ifndef NDEBUG
 716   ReliableQuicStream* stream = GetStream(id);
 717   if (stream != nullptr) {
 718     LOG_IF(DFATAL, priority != stream->EffectivePriority())
 719         << ENDPOINT << "Stream " << id
 720         << "Priorities do not match.  Got: " << priority
 721         << " Expected: " << stream->EffectivePriority();
 722   } else {
 723     LOG(DFATAL) << "Marking unknown stream " << id << " blocked.";
 724   }
 725 #endif
 726
 727   if (id == kCryptoStreamId) {
 728     DCHECK(!has_pending_handshake_);
 729     has_pending_handshake_ = true;
 730     // TODO(jar): Be sure to use the highest priority for the crypto stream,
 731     // perhaps by adding a "special" priority for it that is higher than
 732     // kHighestPriority.
 733     priority = kHighestPriority;
 734   }
 735   write_blocked_streams_.PushBack(id, priority);
 736 }
 737
 738 bool QuicSession::HasDataToWrite() const {
 739   return write_blocked_streams_.HasWriteBlockedCryptoOrHeadersStream() ||
 740          write_blocked_streams_.HasWriteBlockedDataStreams() ||
 741          connection_->HasQueuedData();
 742 }
 743
 744 bool QuicSession::GetSSLInfo(SSLInfo* ssl_info) const {
 745   NOTIMPLEMENTED();
 746   return false;
 747 }
 748
 749 void QuicSession::PostProcessAfterData() {
 750   STLDeleteElements(&closed_streams_);
 751   closed_streams_.clear();
 752
 753   if (connection()->connected() &&
 754       locally_closed_streams_highest_offset_.size() > max_open_streams_) {
 755     // A buggy client may fail to send FIN/RSTs. Don't tolerate this.
 756     connection_->SendConnectionClose(QUIC_TOO_MANY_UNFINISHED_STREAMS);
 757   }
 758 }
 759
 760 void QuicSession::OnSuccessfulVersionNegotiation(const QuicVersion& version) {
 761   // Disable stream level flow control based on negotiated version. Streams may
 762   // have been created with a different version.
 763   if (version < QUIC_VERSION_21) {
 764     GetCryptoStream()->flow_controller()->Disable();
 765     headers_stream_->flow_controller()->Disable();
 766   }
 767 }
 768
 769 bool QuicSession::IsConnectionFlowControlBlocked() const {
 770   return flow_controller_->IsBlocked();
 771 }
 772
 773 bool QuicSession::IsStreamFlowControlBlocked() {
 774   if (headers_stream_->flow_controller()->IsBlocked() ||
 775       GetCryptoStream()->flow_controller()->IsBlocked()) {
 776     return true;
 777   }
 778   for (DataStreamMap::iterator it = stream_map_.begin();
 779        it != stream_map_.end(); ++it) {
 780     if (it->second->flow_controller()->IsBlocked()) {
 781       return true;
 782     }
 783   }
 784   return false;
 785 }
 786
 787 }  // namespace net