Landing Recent QUIC changes until 03:18 AM, May 28, UTC

net/quic/core/quic_session.cc

 // Copyright (c) 2012 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 #include "net/quic/core/quic_session.h"
 
 #include <cstdint>
 #include <utility>
 
 #include "net/quic/core/quic_connection.h"
@@ skipping 27 matching lines @@
       num_locally_closed_incoming_streams_highest_offset_(0),
       error_(QUIC_NO_ERROR),
       flow_controller_(connection_,
                        0,
                        perspective(),
                        kMinimumFlowControlSendWindow,
                        config_.GetInitialSessionFlowControlWindowToSend(),
                        perspective() == Perspective::IS_SERVER,
                        nullptr),
       currently_writing_stream_id_(0),
-      respect_goaway_(true) {}
+      respect_goaway_(true),
+      use_stream_notifier_(
+          FLAGS_quic_reloadable_flag_quic_use_stream_notifier) {}
 
 void QuicSession::Initialize() {
   connection_->set_visitor(this);
+  if (use_stream_notifier_) {
+    connection_->SetStreamNotifier(this);
+  }
   connection_->SetFromConfig(config_);
 
   DCHECK_EQ(kCryptoStreamId, GetMutableCryptoStream()->id());
   static_stream_map_[kCryptoStreamId] = GetMutableCryptoStream();
 }
 
 QuicSession::~QuicSession() {
   QUIC_LOG_IF(WARNING, num_locally_closed_incoming_streams_highest_offset() >
                            max_open_incoming_streams_)
       << "Surprisingly high number of locally closed peer initiated streams"
          "still waiting for final byte offset: "
       << num_locally_closed_incoming_streams_highest_offset();
   QUIC_LOG_IF(WARNING, GetNumLocallyClosedOutgoingStreamsHighestOffset() >
                            max_open_outgoing_streams_)
       << "Surprisingly high number of locally closed self initiated streams"
          "still waiting for final byte offset: "
       << GetNumLocallyClosedOutgoingStreamsHighestOffset();
+  QUIC_LOG_IF(WARNING, !zombie_streams_.empty()) << "Still have zombie streams";
 }
 
 void QuicSession::OnStreamFrame(const QuicStreamFrame& frame) {
   // TODO(rch) deal with the error case of stream id 0.
   QuicStreamId stream_id = frame.stream_id;
   QuicStream* stream = GetOrCreateStream(stream_id);
   if (!stream) {
     // The stream no longer exists, but we may still be interested in the
     // final stream byte offset sent by the peer. A frame with a FIN can give
     // us this offset.
@@ skipping 43 matching lines @@
     DynamicStreamMap::iterator it = dynamic_stream_map_.begin();
     QuicStreamId id = it->first;
     it->second->OnConnectionClosed(error, source);
     // The stream should call CloseStream as part of OnConnectionClosed.
     if (dynamic_stream_map_.find(id) != dynamic_stream_map_.end()) {
       QUIC_BUG << ENDPOINT << "Stream failed to close under OnConnectionClosed";
       CloseStream(id);
     }
   }
 
+  // Cleanup zombie stream map on connection close.
+  while (!zombie_streams_.empty()) {
+    ZombieStreamMap::iterator it = zombie_streams_.begin();
+    closed_streams_.push_back(std::move(it->second));
+    zombie_streams_.erase(it);
+  }
+
   if (visitor_) {
     visitor_->OnConnectionClosed(connection_->connection_id(), error,
                                  error_details);
   }
 }
 
 void QuicSession::OnWriteBlocked() {
   if (visitor_) {
     visitor_->OnWriteBlocked(connection_);
   }
@@ skipping 220 matching lines @@
     QUIC_DLOG(INFO) << ENDPOINT << "Stream is already closed: " << stream_id;
     return;
   }
   QuicStream* stream = it->second.get();
 
   // Tell the stream that a RST has been sent.
   if (locally_reset) {
     stream->set_rst_sent(true);
   }
 
-  closed_streams_.push_back(std::move(it->second));
+  if (stream->IsWaitingForAcks()) {
+    zombie_streams_[stream->id()] = std::move(it->second);
+  } else {
+    closed_streams_.push_back(std::move(it->second));
+  }
 
   // If we haven't received a FIN or RST for this stream, we need to keep track
   // of the how many bytes the stream's flow controller believes it has
   // received, for accurate connection level flow control accounting.
   if (!stream->HasFinalReceivedByteOffset()) {
     InsertLocallyClosedStreamsHighestOffset(
         stream_id, stream->flow_controller()->highest_received_byte_offset());
   }
 
   dynamic_stream_map_.erase(it);
@@ skipping 554 matching lines @@
   return CreateAndActivateStream(GetNextOutgoingStreamId());
 }
 
 QuicStream* QuicSession::CreateAndActivateStream(QuicStreamId id) {
   std::unique_ptr<QuicStream> stream = CreateStream(id);
   QuicStream* stream_ptr = stream.get();
   ActivateStream(std::move(stream));
   return stream_ptr;
 }
 
+void QuicSession::OnStreamDoneWaitingForAcks(QuicStreamId id) {
+  auto it = zombie_streams_.find(id);
+  if (it == zombie_streams_.end()) {
+    return;
+  }
+
+  closed_streams_.push_back(std::move(it->second));
+  zombie_streams_.erase(it);
+}
+
+QuicStream* QuicSession::GetStream(QuicStreamId id) const {
+  auto static_stream = static_stream_map_.find(id);
+  if (static_stream != static_stream_map_.end()) {
+    return static_stream->second;
+  }
+  auto active_stream = dynamic_stream_map_.find(id);
+  if (active_stream != dynamic_stream_map_.end()) {
+    return active_stream->second.get();
+  }
+  auto zombie_stream = zombie_streams_.find(id);
+  if (zombie_stream != zombie_streams_.end()) {
+    return zombie_stream->second.get();
+  }
+  return nullptr;
+}
+
+void QuicSession::OnStreamFrameAcked(const QuicStreamFrame& frame,
+                                     QuicTime::Delta ack_delay_time) {
+  QuicStream* stream = GetStream(frame.stream_id);
+  if (stream != nullptr) {
+    stream->OnStreamFrameAcked(frame, ack_delay_time);
+  }
+}
+
+void QuicSession::OnStreamFrameRetransmitted(const QuicStreamFrame& frame) {
+  QuicStream* stream = GetStream(frame.stream_id);
+  if (stream != nullptr) {
+    stream->OnStreamFrameRetransmitted(frame);
+  }
+}
+
 }  // namespace net