Refactor QuicSession to allow subclasses to do their own thing on pending streams. gfe-relnote: code refactor and v99 only. Not used in production. PiperOrigin-RevId: 248015584 Change-Id: Ie0ac855070b304bd90a8e68392cff3cf4c5ac52a
diff --git a/quic/core/http/quic_spdy_session.cc b/quic/core/http/quic_spdy_session.cc index 163b890..1bcf9cc 100644 --- a/quic/core/http/quic_spdy_session.cc +++ b/quic/core/http/quic_spdy_session.cc
@@ -558,9 +558,9 @@ return GetNumOpenDynamicStreams() > 0; } -bool QuicSpdySession::ShouldBufferIncomingStream(QuicStreamId id) const { - DCHECK_EQ(QUIC_VERSION_99, connection()->transport_version()); - return !QuicUtils::IsBidirectionalStreamId(id); +bool QuicSpdySession::UsesPendingStreams() const { + DCHECK(VersionHasControlStreams(connection()->transport_version())); + return true; } size_t QuicSpdySession::WriteHeadersOnHeadersStreamImpl( @@ -724,7 +724,7 @@ return false; } -void QuicSpdySession::ProcessPendingStreamType(PendingStream* pending) { +void QuicSpdySession::ProcessPendingStream(PendingStream* pending) { DCHECK(VersionHasControlStreams(connection()->transport_version())); struct iovec iov; if (!pending->sequencer()->GetReadableRegion(&iov)) {
diff --git a/quic/core/http/quic_spdy_session.h b/quic/core/http/quic_spdy_session.h index 46f2b8a..65fa346 100644 --- a/quic/core/http/quic_spdy_session.h +++ b/quic/core/http/quic_spdy_session.h
@@ -195,12 +195,12 @@ // Returns true if there are open HTTP requests. bool ShouldKeepConnectionAlive() const override; - // Overridden to buffer incoming streams for version 99. - bool ShouldBufferIncomingStream(QuicStreamId id) const override; + // Overridden to buffer incoming unidirectional streams for version 99. + bool UsesPendingStreams() const override; // Overridden to Process HTTP/3 stream types. No action will be taken if // stream type cannot be read. - void ProcessPendingStreamType(PendingStream* pending) override; + void ProcessPendingStream(PendingStream* pending) override; size_t WriteHeadersOnHeadersStreamImpl( QuicStreamId id,
diff --git a/quic/core/http/quic_spdy_session_test.cc b/quic/core/http/quic_spdy_session_test.cc index 1059a94..347a5f7 100644 --- a/quic/core/http/quic_spdy_session_test.cc +++ b/quic/core/http/quic_spdy_session_test.cc
@@ -267,8 +267,8 @@ using QuicSession::closed_streams; using QuicSession::zombie_streams; - using QuicSpdySession::ProcessPendingStreamType; - using QuicSpdySession::ShouldBufferIncomingStream; + using QuicSpdySession::ProcessPendingStream; + using QuicSpdySession::UsesPendingStreams; private: StrictMock<TestCryptoStream> crypto_stream_; @@ -417,22 +417,11 @@ QuicSpdySessionTestServer, ::testing::ValuesIn(AllSupportedVersions())); -TEST_P(QuicSpdySessionTestServer, ShouldBufferIncomingStreamUnidirectional) { - if (!IsVersion99()) { +TEST_P(QuicSpdySessionTestServer, UsesPendingStreams) { + if (!VersionHasControlStreams(transport_version())) { return; } - EXPECT_TRUE(session_.ShouldBufferIncomingStream( - QuicUtils::GetFirstUnidirectionalStreamId( - connection_->transport_version(), Perspective::IS_CLIENT))); -} - -TEST_P(QuicSpdySessionTestServer, ShouldBufferIncomingStreamBidirectional) { - if (!IsVersion99()) { - return; - } - EXPECT_FALSE(session_.ShouldBufferIncomingStream( - QuicUtils::GetFirstBidirectionalStreamId(connection_->transport_version(), - Perspective::IS_CLIENT))); + EXPECT_TRUE(session_.UsesPendingStreams()); } TEST_P(QuicSpdySessionTestServer, PeerAddress) { @@ -1609,6 +1598,13 @@ QuicSpdySessionTestClient, ::testing::ValuesIn(AllSupportedVersions())); +TEST_P(QuicSpdySessionTestClient, UsesPendingStreams) { + if (!VersionHasControlStreams(transport_version())) { + return; + } + EXPECT_TRUE(session_.UsesPendingStreams()); +} + TEST_P(QuicSpdySessionTestClient, AvailableStreamsClient) { ASSERT_TRUE(session_.GetOrCreateDynamicStream( GetNthServerInitiatedBidirectionalId(2)) != nullptr); @@ -1889,7 +1885,7 @@ // A stop sending frame will be sent to indicate unknown type. EXPECT_CALL(*connection_, SendControlFrame(_)); - session_.ProcessPendingStreamType(&pending); + session_.ProcessPendingStream(&pending); } TEST_P(QuicSpdySessionTestServer, SimplePendingStreamTypeOutOfOrderDelivery) { @@ -1905,13 +1901,13 @@ 'a', 'b', 'c'}; QuicStreamFrame data1(pending.id(), true, 1, QuicStringPiece(&input[1], 3)); pending.OnStreamFrame(data1); - session_.ProcessPendingStreamType(&pending); + session_.ProcessPendingStream(&pending); QuicStreamFrame data2(pending.id(), false, 0, QuicStringPiece(input, 1)); pending.OnStreamFrame(data2); EXPECT_CALL(*connection_, SendControlFrame(_)); - session_.ProcessPendingStreamType(&pending); + session_.ProcessPendingStream(&pending); } TEST_P(QuicSpdySessionTestServer, @@ -1929,17 +1925,17 @@ QuicStreamFrame data1(pending.id(), true, 2, QuicStringPiece(&input[2], 3)); pending.OnStreamFrame(data1); - session_.ProcessPendingStreamType(&pending); + session_.ProcessPendingStream(&pending); QuicStreamFrame data2(pending.id(), false, 0, QuicStringPiece(input, 1)); pending.OnStreamFrame(data2); - session_.ProcessPendingStreamType(&pending); + session_.ProcessPendingStream(&pending); QuicStreamFrame data3(pending.id(), false, 1, QuicStringPiece(&input[1], 1)); pending.OnStreamFrame(data3); EXPECT_CALL(*connection_, SendControlFrame(_)); - session_.ProcessPendingStreamType(&pending); + session_.ProcessPendingStream(&pending); } } // namespace
diff --git a/quic/core/quic_session.cc b/quic/core/quic_session.cc index 8bdf3b5..0545fa6 100644 --- a/quic/core/quic_session.cc +++ b/quic/core/quic_session.cc
@@ -149,6 +149,24 @@ } } +void QuicSession::PendingStreamOnStreamFrame(const QuicStreamFrame& frame) { + DCHECK(VersionHasControlStreams(connection()->transport_version())); + QuicStreamId stream_id = frame.stream_id; + + PendingStream* pending = GetOrCreatePendingStream(stream_id); + + if (!pending) { + if (frame.fin) { + QuicStreamOffset final_byte_offset = frame.offset + frame.data_length; + OnFinalByteOffsetReceived(stream_id, final_byte_offset); + } + return; + } + + pending->OnStreamFrame(frame); + ProcessPendingStream(pending); +} + void QuicSession::OnStreamFrame(const QuicStreamFrame& frame) { // TODO(rch) deal with the error case of stream id 0. QuicStreamId stream_id = frame.stream_id; @@ -167,12 +185,18 @@ return; } - StreamHandler handler = GetOrCreateStreamImpl(stream_id, frame.offset != 0); - if (handler.is_pending) { - handler.pending->OnStreamFrame(frame); + if (VersionHasControlStreams(connection()->transport_version()) && + UsesPendingStreams() && + QuicUtils::GetStreamType(stream_id, perspective(), + IsIncomingStream(stream_id)) == + READ_UNIDIRECTIONAL && + dynamic_stream_map_.find(stream_id) == dynamic_stream_map_.end()) { + PendingStreamOnStreamFrame(frame); return; } + StreamHandler handler = GetOrCreateStreamImpl(stream_id); + if (!handler.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 @@ -296,6 +320,21 @@ return true; } +void QuicSession::PendingStreamOnRstStream(const QuicRstStreamFrame& frame) { + DCHECK(VersionHasControlStreams(connection()->transport_version())); + QuicStreamId stream_id = frame.stream_id; + + PendingStream* pending = GetOrCreatePendingStream(stream_id); + + if (!pending) { + HandleRstOnValidNonexistentStream(frame); + return; + } + + pending->OnRstStreamFrame(frame); + ClosePendingStream(stream_id); +} + void QuicSession::OnRstStream(const QuicRstStreamFrame& frame) { QuicStreamId stream_id = frame.stream_id; if (stream_id == @@ -317,15 +356,18 @@ visitor_->OnRstStreamReceived(frame); } - // may_buffer is true here to allow subclasses to buffer streams until the - // first byte of payload arrives which would allow sessions to delay - // creation of the stream until the type is known. - StreamHandler handler = GetOrCreateStreamImpl(stream_id, /*may_buffer=*/true); - if (handler.is_pending) { - handler.pending->OnRstStreamFrame(frame); - ClosePendingStream(stream_id); + if (VersionHasControlStreams(connection()->transport_version()) && + UsesPendingStreams() && + QuicUtils::GetStreamType(stream_id, perspective(), + IsIncomingStream(stream_id)) == + READ_UNIDIRECTIONAL && + dynamic_stream_map_.find(stream_id) == dynamic_stream_map_.end()) { + PendingStreamOnRstStream(frame); return; } + + StreamHandler handler = GetOrCreateStreamImpl(stream_id); + if (!handler.stream) { HandleRstOnValidNonexistentStream(frame); return; // Errors are handled by GetOrCreateStream. @@ -1200,15 +1242,13 @@ } QuicStream* QuicSession::GetOrCreateStream(const QuicStreamId stream_id) { - StreamHandler handler = - GetOrCreateStreamImpl(stream_id, /*may_buffer=*/false); + StreamHandler handler = GetOrCreateStreamImpl(stream_id); DCHECK(!handler.is_pending); return handler.stream; } QuicSession::StreamHandler QuicSession::GetOrCreateStreamImpl( - QuicStreamId stream_id, - bool may_buffer) { + QuicStreamId stream_id) { if (eliminate_static_stream_map_ && stream_id == QuicUtils::GetCryptoStreamId(connection_->transport_version())) { @@ -1219,7 +1259,7 @@ if (it != static_stream_map_.end()) { return StreamHandler(it->second); } - return GetOrCreateDynamicStreamImpl(stream_id, may_buffer); + return GetOrCreateDynamicStreamImpl(stream_id); } void QuicSession::StreamDraining(QuicStreamId stream_id) { @@ -1254,17 +1294,32 @@ return write_blocked_streams()->ShouldYield(stream_id); } +PendingStream* QuicSession::GetOrCreatePendingStream(QuicStreamId stream_id) { + auto it = pending_stream_map_.find(stream_id); + if (it != pending_stream_map_.end()) { + return it->second.get(); + } + + if (IsClosedStream(stream_id) || + !MaybeIncreaseLargestPeerStreamId(stream_id)) { + return nullptr; + } + + auto pending = QuicMakeUnique<PendingStream>(stream_id, this); + PendingStream* unowned_pending = pending.get(); + pending_stream_map_[stream_id] = std::move(pending); + return unowned_pending; +} + QuicStream* QuicSession::GetOrCreateDynamicStream( const QuicStreamId stream_id) { - StreamHandler handler = - GetOrCreateDynamicStreamImpl(stream_id, /*may_buffer=*/false); + StreamHandler handler = GetOrCreateDynamicStreamImpl(stream_id); DCHECK(!handler.is_pending); return handler.stream; } QuicSession::StreamHandler QuicSession::GetOrCreateDynamicStreamImpl( - QuicStreamId stream_id, - bool may_buffer) { + QuicStreamId stream_id) { DCHECK(!QuicContainsKey(static_stream_map_, stream_id)) << "Attempt to call GetOrCreateDynamicStream for a static stream"; @@ -1282,21 +1337,6 @@ return StreamHandler(); } - auto pending_it = pending_stream_map_.find(stream_id); - if (pending_it != pending_stream_map_.end()) { - DCHECK_EQ(QUIC_VERSION_99, connection_->transport_version()); - if (may_buffer) { - return StreamHandler(pending_it->second.get()); - } - // The stream limit accounting has already been taken care of - // when the PendingStream was created, so there is no need to - // do so here. Now we can create the actual stream from the - // PendingStream. - StreamHandler handler(CreateIncomingStream(std::move(*pending_it->second))); - pending_stream_map_.erase(pending_it); - return handler; - } - // TODO(fkastenholz): If we are creating a new stream and we have // sent a goaway, we should ignore the stream creation. Need to // add code to A) test if goaway was sent ("if (goaway_sent_)") and @@ -1317,18 +1357,6 @@ } } - if (connection_->transport_version() == QUIC_VERSION_99 && may_buffer && - ShouldBufferIncomingStream(stream_id)) { - ++num_dynamic_incoming_streams_; - // Since STREAM frames may arrive out of order, delay creating the - // stream object until the first byte arrives. Buffer the frames and - // handle flow control accounting in the PendingStream. - auto pending = QuicMakeUnique<PendingStream>(stream_id, this); - StreamHandler handler(pending.get()); - pending_stream_map_[stream_id] = std::move(pending); - return handler; - } - return StreamHandler(CreateIncomingStream(stream_id)); }
diff --git a/quic/core/quic_session.h b/quic/core/quic_session.h index e0f2458..f16477c 100644 --- a/quic/core/quic_session.h +++ b/quic/core/quic_session.h
@@ -483,11 +483,11 @@ virtual void OnFinalByteOffsetReceived(QuicStreamId id, QuicStreamOffset final_byte_offset); - // Returns true if incoming streams should be buffered until the first - // byte of the stream arrives. - virtual bool ShouldBufferIncomingStream(QuicStreamId id) const { - return false; - } + // Returns true if incoming unidirectional streams should be buffered until + // the first byte of the stream arrives. + // If a subclass returns true here, it should make sure to implement + // ProcessPendingStream(). + virtual bool UsesPendingStreams() const { return false; } // Register (|id|, |stream|) with the static stream map. Override previous // registrations with the same id. @@ -573,11 +573,11 @@ }; }; - StreamHandler GetOrCreateStreamImpl(QuicStreamId stream_id, bool may_buffer); + StreamHandler GetOrCreateStreamImpl(QuicStreamId stream_id); // Processes the stream type information of |pending| depending on - // different kinds of sessions's own rules. - virtual void ProcessPendingStreamType(PendingStream* pending) {} + // different kinds of sessions' own rules. + virtual void ProcessPendingStream(PendingStream* pending) {} bool eliminate_static_stream_map() const { return eliminate_static_stream_map_; @@ -613,8 +613,9 @@ // closed. QuicStream* GetStream(QuicStreamId id) const; - StreamHandler GetOrCreateDynamicStreamImpl(QuicStreamId stream_id, - bool may_buffer); + StreamHandler GetOrCreateDynamicStreamImpl(QuicStreamId stream_id); + + PendingStream* GetOrCreatePendingStream(QuicStreamId stream_id); // Let streams and control frame managers retransmit lost data, returns true // if all lost data is retransmitted. Returns false otherwise. @@ -623,6 +624,14 @@ // Closes the pending stream |stream_id| before it has been created. void ClosePendingStream(QuicStreamId stream_id); + // Creates or gets pending stream, feeds it with |frame|, and processes the + // pending stream. + void PendingStreamOnStreamFrame(const QuicStreamFrame& frame); + + // Creates or gets pending strea, feed it with |frame|, and closes the pending + // stream. + void PendingStreamOnRstStream(const QuicRstStreamFrame& frame); + // Keep track of highest received byte offset of locally closed streams, while // waiting for a definitive final highest offset from the peer. std::map<QuicStreamId, QuicStreamOffset>
diff --git a/quic/core/quic_session_test.cc b/quic/core/quic_session_test.cc index a441dc4..8f418c0 100644 --- a/quic/core/quic_session_test.cc +++ b/quic/core/quic_session_test.cc
@@ -138,7 +138,7 @@ CurrentSupportedVersions()), crypto_stream_(this), writev_consumes_all_data_(false), - should_buffer_incoming_streams_(false), + uses_pending_streams_(false), num_incoming_streams_created_(0) { Initialize(); this->connection()->SetEncrypter( @@ -209,6 +209,17 @@ return stream; } + // QuicSession doesn't do anything in this method. So it's overridden here to + // test that the session handles pending streams correctly in terms of + // receiving stream frames. + void ProcessPendingStream(PendingStream* pending) override { + struct iovec iov; + if (pending->sequencer()->GetReadableRegion(&iov)) { + // Create TestStream once the first byte is received. + CreateIncomingStream(std::move(*pending)); + } + } + bool IsClosedStream(QuicStreamId id) { return QuicSession::IsClosedStream(id); } @@ -279,12 +290,10 @@ return WritevData(stream, stream->id(), bytes, 0, FIN); } - bool ShouldBufferIncomingStream(QuicStreamId id) const override { - return should_buffer_incoming_streams_; - } + bool UsesPendingStreams() const override { return uses_pending_streams_; } - void set_should_buffer_incoming_streams(bool should_buffer_incoming_streams) { - should_buffer_incoming_streams_ = should_buffer_incoming_streams; + void set_uses_pending_streams(bool uses_pending_streams) { + uses_pending_streams_ = uses_pending_streams; } int num_incoming_streams_created() const { @@ -299,7 +308,7 @@ StrictMock<TestCryptoStream> crypto_stream_; bool writev_consumes_all_data_; - bool should_buffer_incoming_streams_; + bool uses_pending_streams_; QuicFrame save_frame_; int num_incoming_streams_created_; }; @@ -1575,7 +1584,7 @@ } TEST_P(QuicSessionTestServer, NoPendingStreams) { - session_.set_should_buffer_incoming_streams(false); + session_.set_uses_pending_streams(false); QuicStreamId stream_id = QuicUtils::GetFirstUnidirectionalStreamId( transport_version(), Perspective::IS_CLIENT); @@ -1592,7 +1601,7 @@ if (connection_->transport_version() != QUIC_VERSION_99) { return; } - session_.set_should_buffer_incoming_streams(true); + session_.set_uses_pending_streams(true); QuicStreamId stream_id = QuicUtils::GetFirstUnidirectionalStreamId( transport_version(), Perspective::IS_CLIENT); @@ -1609,7 +1618,7 @@ if (connection_->transport_version() != QUIC_VERSION_99) { return; } - session_.set_should_buffer_incoming_streams(true); + session_.set_uses_pending_streams(true); QuicStreamId stream_id = QuicUtils::GetFirstUnidirectionalStreamId( transport_version(), Perspective::IS_CLIENT);