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::vector;
  20
  21 namespace net {
  22
  23 #define ENDPOINT (is_server() ? "Server: " : " Client: ")
  24
  25 // We want to make sure we delete any closed streams in a safe manner.
  26 // To avoid deleting a stream in mid-operation, we have a simple shim between
  27 // us and the stream, so we can delete any streams when we return from
  28 // processing.
  29 //
  30 // We could just override the base methods, but this makes it easier to make
  31 // sure we don't miss any.
  32 class VisitorShim : public QuicConnectionVisitorInterface {
  33  public:
  34   explicit VisitorShim(QuicSession* session) : session_(session) {}
  35
  36   virtual void OnStreamFrames(const vector<QuicStreamFrame>& frames) OVERRIDE {
  37     session_->OnStreamFrames(frames);
  38     session_->PostProcessAfterData();
  39   }
  40   virtual void OnRstStream(const QuicRstStreamFrame& frame) OVERRIDE {
  41     session_->OnRstStream(frame);
  42     session_->PostProcessAfterData();
  43   }
  44
  45   virtual void OnGoAway(const QuicGoAwayFrame& frame) OVERRIDE {
  46     session_->OnGoAway(frame);
  47     session_->PostProcessAfterData();
  48   }
  49
  50   virtual void OnWindowUpdateFrames(const vector<QuicWindowUpdateFrame>& frames)
  51       OVERRIDE {
  52     session_->OnWindowUpdateFrames(frames);
  53     session_->PostProcessAfterData();
  54   }
  55
  56   virtual void OnBlockedFrames(const vector<QuicBlockedFrame>& frames)
  57       OVERRIDE {
  58     session_->OnBlockedFrames(frames);
  59     session_->PostProcessAfterData();
  60   }
  61
  62   virtual void OnCanWrite() OVERRIDE {
  63     session_->OnCanWrite();
  64     session_->PostProcessAfterData();
  65   }
  66
  67   virtual void OnSuccessfulVersionNegotiation(
  68       const QuicVersion& version) OVERRIDE {
  69     session_->OnSuccessfulVersionNegotiation(version);
  70   }
  71
  72   virtual void OnConnectionClosed(
  73       QuicErrorCode error, bool from_peer) OVERRIDE {
  74     session_->OnConnectionClosed(error, from_peer);
  75     // The session will go away, so don't bother with cleanup.
  76   }
  77
  78   virtual void OnWriteBlocked() OVERRIDE {
  79     session_->OnWriteBlocked();
  80   }
  81
  82   virtual bool WillingAndAbleToWrite() const OVERRIDE {
  83     return session_->WillingAndAbleToWrite();
  84   }
  85
  86   virtual bool HasPendingHandshake() const OVERRIDE {
  87     return session_->HasPendingHandshake();
  88   }
  89
  90   virtual 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_.max_streams_per_connection()),
 103       next_stream_id_(is_server() ? 2 : 3),
 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   connection_->set_visitor(visitor_shim_.get());
 122   connection_->SetFromConfig(config_);
 123   if (connection_->connected()) {
 124     connection_->SetOverallConnectionTimeout(
 125         config_.max_time_before_crypto_handshake());
 126   }
 127   headers_stream_.reset(new QuicHeadersStream(this));
 128   if (!is_server()) {
 129     // For version above QUIC v12, the headers stream is stream 3, so the
 130     // next available local stream ID should be 5.
 131     DCHECK_EQ(kHeadersStreamId, next_stream_id_);
 132     next_stream_id_ += 2;
 133   }
 134 }
 135
 136 QuicSession::~QuicSession() {
 137   STLDeleteElements(&closed_streams_);
 138   STLDeleteValues(&stream_map_);
 139
 140   DLOG_IF(WARNING,
 141           locally_closed_streams_highest_offset_.size() > max_open_streams_)
 142       << "Surprisingly high number of locally closed streams still waiting for "
 143          "final byte offset: " << locally_closed_streams_highest_offset_.size();
 144 }
 145
 146 void QuicSession::OnStreamFrames(const vector<QuicStreamFrame>& frames) {
 147   for (size_t i = 0; i < frames.size(); ++i) {
 148     // TODO(rch) deal with the error case of stream id 0.
 149     const QuicStreamFrame& frame = frames[i];
 150     QuicStreamId stream_id = frame.stream_id;
 151     ReliableQuicStream* stream = GetStream(stream_id);
 152     if (!stream) {
 153       // The stream no longer exists, but we may still be interested in the
 154       // final stream byte offset sent by the peer. A frame with a FIN can give
 155       // us this offset.
 156       if (frame.fin) {
 157         QuicStreamOffset final_byte_offset =
 158             frame.offset + frame.data.TotalBufferSize();
 159         UpdateFlowControlOnFinalReceivedByteOffset(stream_id,
 160                                                    final_byte_offset);
 161       }
 162
 163       continue;
 164     }
 165     stream->OnStreamFrame(frames[i]);
 166   }
 167 }
 168
 169 void QuicSession::OnStreamHeaders(QuicStreamId stream_id,
 170                                   StringPiece headers_data) {
 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->OnStreamHeaders(headers_data);
 177 }
 178
 179 void QuicSession::OnStreamHeadersPriority(QuicStreamId stream_id,
 180                                           QuicPriority priority) {
 181   QuicDataStream* stream = GetDataStream(stream_id);
 182   if (!stream) {
 183     // It's quite possible to receive headers after a stream has been reset.
 184     return;
 185   }
 186   stream->OnStreamHeadersPriority(priority);
 187 }
 188
 189 void QuicSession::OnStreamHeadersComplete(QuicStreamId stream_id,
 190                                           bool fin,
 191                                           size_t frame_len) {
 192   QuicDataStream* stream = GetDataStream(stream_id);
 193   if (!stream) {
 194     // It's quite possible to receive headers after a stream has been reset.
 195     return;
 196   }
 197   stream->OnStreamHeadersComplete(fin, frame_len);
 198 }
 199
 200 void QuicSession::OnRstStream(const QuicRstStreamFrame& frame) {
 201   if (frame.stream_id == kCryptoStreamId) {
 202     connection()->SendConnectionCloseWithDetails(
 203         QUIC_INVALID_STREAM_ID,
 204         "Attempt to reset the crypto stream");
 205     return;
 206   }
 207   if (frame.stream_id == kHeadersStreamId) {
 208     connection()->SendConnectionCloseWithDetails(
 209         QUIC_INVALID_STREAM_ID,
 210         "Attempt to reset the headers stream");
 211     return;
 212   }
 213
 214   QuicDataStream* stream = GetDataStream(frame.stream_id);
 215   if (!stream) {
 216     // The RST frame contains the final byte offset for the stream: we can now
 217     // update the connection level flow controller if needed.
 218     UpdateFlowControlOnFinalReceivedByteOffset(frame.stream_id,
 219                                                frame.byte_offset);
 220     return;  // Errors are handled by GetStream.
 221   }
 222
 223   stream->OnStreamReset(frame);
 224 }
 225
 226 void QuicSession::OnGoAway(const QuicGoAwayFrame& frame) {
 227   DCHECK(frame.last_good_stream_id < next_stream_id_);
 228   goaway_received_ = true;
 229 }
 230
 231 void QuicSession::OnConnectionClosed(QuicErrorCode error, bool from_peer) {
 232   DCHECK(!connection_->connected());
 233   if (error_ == QUIC_NO_ERROR) {
 234     error_ = error;
 235   }
 236
 237   while (!stream_map_.empty()) {
 238     DataStreamMap::iterator it = stream_map_.begin();
 239     QuicStreamId id = it->first;
 240     it->second->OnConnectionClosed(error, from_peer);
 241     // The stream should call CloseStream as part of OnConnectionClosed.
 242     if (stream_map_.find(id) != stream_map_.end()) {
 243       LOG(DFATAL) << ENDPOINT
 244                   << "Stream failed to close under OnConnectionClosed";
 245       CloseStream(id);
 246     }
 247   }
 248 }
 249
 250 void QuicSession::OnWindowUpdateFrames(
 251     const vector<QuicWindowUpdateFrame>& frames) {
 252   bool connection_window_updated = false;
 253   for (size_t i = 0; i < frames.size(); ++i) {
 254     // Stream may be closed by the time we receive a WINDOW_UPDATE, so we can't
 255     // assume that it still exists.
 256     QuicStreamId stream_id = frames[i].stream_id;
 257     if (stream_id == kConnectionLevelId) {
 258       // This is a window update that applies to the connection, rather than an
 259       // individual stream.
 260       DVLOG(1) << ENDPOINT
 261                << "Received connection level flow control window update with "
 262                   "byte offset: " << frames[i].byte_offset;
 263       if (FLAGS_enable_quic_connection_flow_control_2 &&
 264           flow_controller_->UpdateSendWindowOffset(frames[i].byte_offset)) {
 265         connection_window_updated = true;
 266       }
 267       continue;
 268     }
 269
 270     if (connection_->version() <= QUIC_VERSION_20 &&
 271         (stream_id == kCryptoStreamId || stream_id == kHeadersStreamId)) {
 272       DLOG(DFATAL) << "WindowUpdate for stream " << stream_id << " in version "
 273                    << QuicVersionToString(connection_->version());
 274       return;
 275     }
 276
 277     ReliableQuicStream* stream = GetStream(stream_id);
 278     if (stream) {
 279       stream->OnWindowUpdateFrame(frames[i]);
 280     }
 281   }
 282
 283   // Connection level flow control window has increased, so blocked streams can
 284   // write again.
 285   if (connection_window_updated) {
 286     OnCanWrite();
 287   }
 288 }
 289
 290 void QuicSession::OnBlockedFrames(const vector<QuicBlockedFrame>& frames) {
 291   for (size_t i = 0; i < frames.size(); ++i) {
 292     // TODO(rjshade): Compare our flow control receive windows for specified
 293     //                streams: if we have a large window then maybe something
 294     //                had gone wrong with the flow control accounting.
 295     DVLOG(1) << ENDPOINT << "Received BLOCKED frame with stream id: "
 296              << frames[i].stream_id;
 297   }
 298 }
 299
 300 void QuicSession::OnCanWrite() {
 301   // We limit the number of writes to the number of pending streams. If more
 302   // streams become pending, WillingAndAbleToWrite will be true, which will
 303   // cause the connection to request resumption before yielding to other
 304   // connections.
 305   size_t num_writes = write_blocked_streams_.NumBlockedStreams();
 306   if (flow_controller_->IsBlocked()) {
 307     // If we are connection level flow control blocked, then only allow the
 308     // crypto and headers streams to try writing as all other streams will be
 309     // blocked.
 310     num_writes = 0;
 311     if (write_blocked_streams_.crypto_stream_blocked()) {
 312       num_writes += 1;
 313     }
 314     if (write_blocked_streams_.headers_stream_blocked()) {
 315       num_writes += 1;
 316     }
 317   }
 318   if (num_writes == 0) {
 319     return;
 320   }
 321
 322   QuicConnection::ScopedPacketBundler ack_bundler(
 323       connection_.get(), QuicConnection::NO_ACK);
 324   for (size_t i = 0; i < num_writes; ++i) {
 325     if (!(write_blocked_streams_.HasWriteBlockedCryptoOrHeadersStream() ||
 326           write_blocked_streams_.HasWriteBlockedDataStreams())) {
 327       // Writing one stream removed another!? Something's broken.
 328       LOG(DFATAL) << "WriteBlockedStream is missing";
 329       connection_->CloseConnection(QUIC_INTERNAL_ERROR, false);
 330       return;
 331     }
 332     if (!connection_->CanWriteStreamData()) {
 333       return;
 334     }
 335     QuicStreamId stream_id = write_blocked_streams_.PopFront();
 336     if (stream_id == kCryptoStreamId) {
 337       has_pending_handshake_ = false;  // We just popped it.
 338     }
 339     ReliableQuicStream* stream = GetStream(stream_id);
 340     if (stream != NULL && !stream->flow_controller()->IsBlocked()) {
 341       // If the stream can't write all bytes, it'll re-add itself to the blocked
 342       // list.
 343       stream->OnCanWrite();
 344     }
 345   }
 346 }
 347
 348 bool QuicSession::WillingAndAbleToWrite() const {
 349   // If the crypto or headers streams are blocked, we want to schedule a write -
 350   // they don't get blocked by connection level flow control. Otherwise only
 351   // schedule a write if we are not flow control blocked at the connection
 352   // level.
 353   return write_blocked_streams_.HasWriteBlockedCryptoOrHeadersStream() ||
 354          (!flow_controller_->IsBlocked() &&
 355           write_blocked_streams_.HasWriteBlockedDataStreams());
 356 }
 357
 358 bool QuicSession::HasPendingHandshake() const {
 359   return has_pending_handshake_;
 360 }
 361
 362 bool QuicSession::HasOpenDataStreams() const {
 363   return GetNumOpenStreams() > 0;
 364 }
 365
 366 QuicConsumedData QuicSession::WritevData(
 367     QuicStreamId id,
 368     const IOVector& data,
 369     QuicStreamOffset offset,
 370     bool fin,
 371     FecProtection fec_protection,
 372     QuicAckNotifier::DelegateInterface* ack_notifier_delegate) {
 373   return connection_->SendStreamData(id, data, offset, fin, fec_protection,
 374                                      ack_notifier_delegate);
 375 }
 376
 377 size_t QuicSession::WriteHeaders(
 378     QuicStreamId id,
 379     const SpdyHeaderBlock& headers,
 380     bool fin,
 381     QuicAckNotifier::DelegateInterface* ack_notifier_delegate) {
 382   return headers_stream_->WriteHeaders(id, headers, fin, ack_notifier_delegate);
 383 }
 384
 385 void QuicSession::SendRstStream(QuicStreamId id,
 386                                 QuicRstStreamErrorCode error,
 387                                 QuicStreamOffset bytes_written) {
 388   if (connection()->connected()) {
 389     // Only send a RST_STREAM frame if still connected.
 390     connection_->SendRstStream(id, error, bytes_written);
 391   }
 392   CloseStreamInner(id, true);
 393 }
 394
 395 void QuicSession::SendGoAway(QuicErrorCode error_code, const string& reason) {
 396   if (goaway_sent_) {
 397     return;
 398   }
 399   goaway_sent_ = true;
 400   connection_->SendGoAway(error_code, largest_peer_created_stream_id_, reason);
 401 }
 402
 403 void QuicSession::CloseStream(QuicStreamId stream_id) {
 404   CloseStreamInner(stream_id, false);
 405 }
 406
 407 void QuicSession::CloseStreamInner(QuicStreamId stream_id,
 408                                    bool locally_reset) {
 409   DVLOG(1) << ENDPOINT << "Closing stream " << stream_id;
 410
 411   DataStreamMap::iterator it = stream_map_.find(stream_id);
 412   if (it == stream_map_.end()) {
 413     DVLOG(1) << ENDPOINT << "Stream is already closed: " << stream_id;
 414     return;
 415   }
 416   QuicDataStream* stream = it->second;
 417
 418   // Tell the stream that a RST has been sent.
 419   if (locally_reset) {
 420     stream->set_rst_sent(true);
 421   }
 422
 423   closed_streams_.push_back(it->second);
 424
 425   // If we haven't received a FIN or RST for this stream, we need to keep track
 426   // of the how many bytes the stream's flow controller believes it has
 427   // received, for accurate connection level flow control accounting.
 428   if (!stream->HasFinalReceivedByteOffset() &&
 429       stream->flow_controller()->IsEnabled() &&
 430       FLAGS_enable_quic_connection_flow_control_2) {
 431     locally_closed_streams_highest_offset_[stream_id] =
 432         stream->flow_controller()->highest_received_byte_offset();
 433   }
 434
 435   stream_map_.erase(it);
 436   stream->OnClose();
 437 }
 438
 439 void QuicSession::UpdateFlowControlOnFinalReceivedByteOffset(
 440     QuicStreamId stream_id, QuicStreamOffset final_byte_offset) {
 441   if (!FLAGS_enable_quic_connection_flow_control_2) {
 442     return;
 443   }
 444
 445   map<QuicStreamId, QuicStreamOffset>::iterator it =
 446       locally_closed_streams_highest_offset_.find(stream_id);
 447   if (it == locally_closed_streams_highest_offset_.end()) {
 448     return;
 449   }
 450
 451   DVLOG(1) << ENDPOINT << "Received final byte offset " << final_byte_offset
 452            << " for stream " << stream_id;
 453   uint64 offset_diff = final_byte_offset - it->second;
 454   if (flow_controller_->UpdateHighestReceivedOffset(
 455       flow_controller_->highest_received_byte_offset() + offset_diff)) {
 456     // If the final offset violates flow control, close the connection now.
 457     if (flow_controller_->FlowControlViolation()) {
 458       connection_->SendConnectionClose(
 459           QUIC_FLOW_CONTROL_RECEIVED_TOO_MUCH_DATA);
 460       return;
 461     }
 462   }
 463
 464   flow_controller_->AddBytesConsumed(offset_diff);
 465   locally_closed_streams_highest_offset_.erase(it);
 466 }
 467
 468 bool QuicSession::IsEncryptionEstablished() {
 469   return GetCryptoStream()->encryption_established();
 470 }
 471
 472 bool QuicSession::IsCryptoHandshakeConfirmed() {
 473   return GetCryptoStream()->handshake_confirmed();
 474 }
 475
 476 void QuicSession::OnConfigNegotiated() {
 477   connection_->SetFromConfig(config_);
 478   QuicVersion version = connection()->version();
 479   if (version <= QUIC_VERSION_16) {
 480     return;
 481   }
 482
 483   if (version <= QUIC_VERSION_19) {
 484     // QUIC_VERSION_17,18,19 don'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_20 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_20) {
 524     GetCryptoStream()->flow_controller()->UpdateSendWindowOffset(new_window);
 525     headers_stream_->flow_controller()->UpdateSendWindowOffset(new_window);
 526   }
 527   for (DataStreamMap::iterator it = stream_map_.begin();
 528        it != stream_map_.end(); ++it) {
 529     it->second->flow_controller()->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(INITIAL_ENCRYPTION_ONLY);
 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       connection_->SetOverallConnectionTimeout(QuicTime::Delta::Infinite());
 567       max_open_streams_ = config_.max_streams_per_connection();
 568       break;
 569
 570     default:
 571       LOG(ERROR) << ENDPOINT << "Got unknown handshake event: " << event;
 572   }
 573 }
 574
 575 void QuicSession::OnCryptoHandshakeMessageSent(
 576     const CryptoHandshakeMessage& message) {
 577 }
 578
 579 void QuicSession::OnCryptoHandshakeMessageReceived(
 580     const CryptoHandshakeMessage& message) {
 581 }
 582
 583 QuicConfig* QuicSession::config() {
 584   return &config_;
 585 }
 586
 587 void QuicSession::ActivateStream(QuicDataStream* stream) {
 588   DVLOG(1) << ENDPOINT << "num_streams: " << stream_map_.size()
 589            << ". activating " << stream->id();
 590   DCHECK_EQ(stream_map_.count(stream->id()), 0u);
 591   stream_map_[stream->id()] = stream;
 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 NULL;
 614   }
 615   if (stream_id == kHeadersStreamId) {
 616     DLOG(FATAL) << "Attempt to call GetDataStream with the headers stream id";
 617     return NULL;
 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 NULL;
 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 NULL;
 636   }
 637
 638   return GetIncomingDataStream(stream_id);
 639 }
 640
 641 QuicDataStream* QuicSession::GetIncomingDataStream(QuicStreamId stream_id) {
 642   if (IsClosedStream(stream_id)) {
 643     return NULL;
 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 NULL;
 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 == NULL) {
 676     return NULL;
 677   }
 678   ActivateStream(stream);
 679   return stream;
 680 }
 681
 682 bool QuicSession::IsClosedStream(QuicStreamId id) {
 683   DCHECK_NE(0u, id);
 684   if (id == kCryptoStreamId) {
 685     return false;
 686   }
 687   if (id == kHeadersStreamId) {
 688     return false;
 689   }
 690   if (ContainsKey(stream_map_, id)) {
 691     // Stream is active
 692     return false;
 693   }
 694   if (id % 2 == next_stream_id_ % 2) {
 695     // Locally created streams are strictly in-order.  If the id is in the
 696     // range of created streams and it's not active, it must have been closed.
 697     return id < next_stream_id_;
 698   }
 699   // For peer created streams, we also need to consider implicitly created
 700   // streams.
 701   return id <= largest_peer_created_stream_id_ &&
 702       implicitly_created_streams_.count(id) == 0;
 703 }
 704
 705 size_t QuicSession::GetNumOpenStreams() const {
 706   return stream_map_.size() + implicitly_created_streams_.size();
 707 }
 708
 709 void QuicSession::MarkWriteBlocked(QuicStreamId id, QuicPriority priority) {
 710 #ifndef NDEBUG
 711   ReliableQuicStream* stream = GetStream(id);
 712   if (stream != NULL) {
 713     LOG_IF(DFATAL, priority != stream->EffectivePriority())
 714         << ENDPOINT << "Stream " << id
 715         << "Priorities do not match.  Got: " << priority
 716         << " Expected: " << stream->EffectivePriority();
 717   } else {
 718     LOG(DFATAL) << "Marking unknown stream " << id << " blocked.";
 719   }
 720 #endif
 721
 722   if (id == kCryptoStreamId) {
 723     DCHECK(!has_pending_handshake_);
 724     has_pending_handshake_ = true;
 725     // TODO(jar): Be sure to use the highest priority for the crypto stream,
 726     // perhaps by adding a "special" priority for it that is higher than
 727     // kHighestPriority.
 728     priority = kHighestPriority;
 729   }
 730   write_blocked_streams_.PushBack(id, priority);
 731 }
 732
 733 bool QuicSession::HasDataToWrite() const {
 734   return write_blocked_streams_.HasWriteBlockedCryptoOrHeadersStream() ||
 735          write_blocked_streams_.HasWriteBlockedDataStreams() ||
 736          connection_->HasQueuedData();
 737 }
 738
 739 bool QuicSession::GetSSLInfo(SSLInfo* ssl_info) const {
 740   NOTIMPLEMENTED();
 741   return false;
 742 }
 743
 744 void QuicSession::PostProcessAfterData() {
 745   STLDeleteElements(&closed_streams_);
 746   closed_streams_.clear();
 747 }
 748
 749 void QuicSession::OnSuccessfulVersionNegotiation(const QuicVersion& version) {
 750   if (version < QUIC_VERSION_19) {
 751     flow_controller_->Disable();
 752   }
 753
 754   // Disable stream level flow control based on negotiated version. Streams may
 755   // have been created with a different version.
 756   if (version <= QUIC_VERSION_20) {
 757     GetCryptoStream()->flow_controller()->Disable();
 758     headers_stream_->flow_controller()->Disable();
 759   }
 760   for (DataStreamMap::iterator it = stream_map_.begin();
 761        it != stream_map_.end(); ++it) {
 762     if (version <= QUIC_VERSION_16) {
 763       it->second->flow_controller()->Disable();
 764     }
 765   }
 766 }
 767
 768 }  // namespace net