Tracks streams with ready metadata or trailers separately from write_scheduler_. PiperOrigin-RevId: 415351912
diff --git a/http2/adapter/oghttp2_adapter_test.cc b/http2/adapter/oghttp2_adapter_test.cc index 6235d34..3b22a92 100644 --- a/http2/adapter/oghttp2_adapter_test.cc +++ b/http2/adapter/oghttp2_adapter_test.cc
@@ -1756,6 +1756,196 @@ EXPECT_FALSE(adapter->want_write()); } +TEST(OgHttp2AdapterClientTest, ClientEncountersFlowControlBlock) { + DataSavingVisitor visitor; + OgHttp2Adapter::Options options{.perspective = Perspective::kClient}; + auto adapter = OgHttp2Adapter::Create(visitor, options); + + testing::InSequence s; + + const std::vector<const Header> headers1 = + ToHeaders({{":method", "GET"}, + {":scheme", "http"}, + {":authority", "example.com"}, + {":path", "/this/is/request/one"}}); + + const std::string kBody = std::string(100 * 1024, 'a'); + auto body1 = absl::make_unique<TestDataFrameSource>(visitor, false); + body1->AppendPayload(kBody); + body1->EndData(); + + const int32_t stream_id1 = + adapter->SubmitRequest(headers1, std::move(body1), nullptr); + ASSERT_GT(stream_id1, 0); + + const std::vector<const Header> headers2 = + ToHeaders({{":method", "GET"}, + {":scheme", "http"}, + {":authority", "example.com"}, + {":path", "/this/is/request/two"}}); + + auto body2 = absl::make_unique<TestDataFrameSource>(visitor, false); + body2->AppendPayload(kBody); + body2->EndData(); + + const int32_t stream_id2 = + adapter->SubmitRequest(headers2, std::move(body2), nullptr); + ASSERT_GT(stream_id2, 0); + + EXPECT_CALL(visitor, OnBeforeFrameSent(SETTINGS, 0, _, 0x0)); + EXPECT_CALL(visitor, OnFrameSent(SETTINGS, 0, _, 0x0, 0)); + EXPECT_CALL(visitor, OnBeforeFrameSent(HEADERS, stream_id1, _, 0x4)); + EXPECT_CALL(visitor, OnFrameSent(HEADERS, stream_id1, _, 0x4, 0)); + EXPECT_CALL(visitor, OnBeforeFrameSent(HEADERS, stream_id2, _, 0x4)); + EXPECT_CALL(visitor, OnFrameSent(HEADERS, stream_id2, _, 0x4, 0)); + // 4 DATA frames should saturate the default 64kB stream/connection flow + // control window. + EXPECT_CALL(visitor, OnFrameSent(DATA, stream_id1, _, 0x0, 0)).Times(4); + + int result = adapter->Send(); + EXPECT_EQ(0, result); + EXPECT_EQ(0, adapter->GetSendWindowSize()); + + const std::string stream_frames = TestFrameSequence() + .ServerPreface() + .WindowUpdate(0, 80000) + .WindowUpdate(stream_id1, 20000) + .Serialize(); + + // Server preface (empty SETTINGS) + EXPECT_CALL(visitor, OnFrameHeader(0, 0, SETTINGS, 0)); + EXPECT_CALL(visitor, OnSettingsStart()); + EXPECT_CALL(visitor, OnSettingsEnd()); + + EXPECT_CALL(visitor, OnFrameHeader(0, 4, WINDOW_UPDATE, 0)); + EXPECT_CALL(visitor, OnWindowUpdate(0, 80000)); + EXPECT_CALL(visitor, OnFrameHeader(1, 4, WINDOW_UPDATE, 0)); + EXPECT_CALL(visitor, OnWindowUpdate(1, 20000)); + + const int64_t stream_result = adapter->ProcessBytes(stream_frames); + EXPECT_EQ(stream_frames.size(), static_cast<size_t>(stream_result)); + + EXPECT_CALL(visitor, OnBeforeFrameSent(SETTINGS, 0, _, 0x1)); + EXPECT_CALL(visitor, OnFrameSent(SETTINGS, 0, _, 0x1, 0)); + + EXPECT_CALL(visitor, OnFrameSent(DATA, stream_id2, _, 0x0, 0)) + .Times(testing::AtLeast(1)); + EXPECT_CALL(visitor, OnFrameSent(DATA, stream_id1, _, 0x0, 0)) + .Times(testing::AtLeast(1)); + + EXPECT_TRUE(adapter->want_write()); + result = adapter->Send(); + EXPECT_EQ(0, result); +} + +TEST(OgHttp2AdapterClientTest, ClientSendsTrailersAfterFlowControlBlock) { + DataSavingVisitor visitor; + OgHttp2Adapter::Options options{.perspective = Perspective::kClient}; + auto adapter = OgHttp2Adapter::Create(visitor, options); + + testing::InSequence s; + + const std::vector<const Header> headers1 = + ToHeaders({{":method", "GET"}, + {":scheme", "http"}, + {":authority", "example.com"}, + {":path", "/this/is/request/one"}}); + + auto body1 = absl::make_unique<TestDataFrameSource>(visitor, false); + body1->AppendPayload("Really small body."); + body1->EndData(); + + const int32_t stream_id1 = + adapter->SubmitRequest(headers1, std::move(body1), nullptr); + ASSERT_GT(stream_id1, 0); + + const std::vector<const Header> headers2 = + ToHeaders({{":method", "GET"}, + {":scheme", "http"}, + {":authority", "example.com"}, + {":path", "/this/is/request/two"}}); + + const std::string kBody = std::string(100 * 1024, 'a'); + auto body2 = absl::make_unique<TestDataFrameSource>(visitor, false); + body2->AppendPayload(kBody); + body2->EndData(); + + const int32_t stream_id2 = + adapter->SubmitRequest(headers2, std::move(body2), nullptr); + ASSERT_GT(stream_id2, 0); + + EXPECT_CALL(visitor, OnBeforeFrameSent(SETTINGS, 0, _, 0x0)); + EXPECT_CALL(visitor, OnFrameSent(SETTINGS, 0, _, 0x0, 0)); + EXPECT_CALL(visitor, OnBeforeFrameSent(HEADERS, stream_id1, _, 0x4)); + EXPECT_CALL(visitor, OnFrameSent(HEADERS, stream_id1, _, 0x4, 0)); + EXPECT_CALL(visitor, OnBeforeFrameSent(HEADERS, stream_id2, _, 0x4)); + EXPECT_CALL(visitor, OnFrameSent(HEADERS, stream_id2, _, 0x4, 0)); + EXPECT_CALL(visitor, OnFrameSent(DATA, stream_id1, _, 0x0, 0)).Times(1); + // 4 DATA frames should saturate the default 64kB stream/connection flow + // control window. + EXPECT_CALL(visitor, OnFrameSent(DATA, stream_id2, _, 0x0, 0)).Times(4); + + int result = adapter->Send(); + EXPECT_EQ(0, result); + EXPECT_FALSE(adapter->want_write()); + EXPECT_EQ(0, adapter->GetSendWindowSize()); + + const std::vector<const Header> trailers1 = + ToHeaders({{"extra-info", "Trailers are weird but good?"}}); + adapter->SubmitTrailer(stream_id1, trailers1); + + EXPECT_CALL(visitor, OnBeforeFrameSent(HEADERS, stream_id1, _, 0x5)); + EXPECT_CALL(visitor, OnFrameSent(HEADERS, stream_id1, _, 0x5, 0)); + + result = adapter->Send(); + EXPECT_EQ(0, result); +} + +TEST(OgHttp2AdapterClientTest, ClientSendsMetadataAfterFlowControlBlock) { + DataSavingVisitor visitor; + OgHttp2Adapter::Options options{.perspective = Perspective::kClient}; + auto adapter = OgHttp2Adapter::Create(visitor, options); + + testing::InSequence s; + + const std::vector<const Header> headers1 = + ToHeaders({{":method", "GET"}, + {":scheme", "http"}, + {":authority", "example.com"}, + {":path", "/this/is/request/one"}}); + + const std::string kBody = std::string(100 * 1024, 'a'); + auto body1 = absl::make_unique<TestDataFrameSource>(visitor, false); + body1->AppendPayload(kBody); + body1->EndData(); + + const int32_t stream_id1 = + adapter->SubmitRequest(headers1, std::move(body1), nullptr); + ASSERT_GT(stream_id1, 0); + + EXPECT_CALL(visitor, OnBeforeFrameSent(SETTINGS, 0, _, 0x0)); + EXPECT_CALL(visitor, OnFrameSent(SETTINGS, 0, _, 0x0, 0)); + EXPECT_CALL(visitor, OnBeforeFrameSent(HEADERS, stream_id1, _, 0x4)); + EXPECT_CALL(visitor, OnFrameSent(HEADERS, stream_id1, _, 0x4, 0)); + // 4 DATA frames should saturate the default 64kB stream/connection flow + // control window. + EXPECT_CALL(visitor, OnFrameSent(DATA, stream_id1, _, 0x0, 0)).Times(4); + + int result = adapter->Send(); + EXPECT_EQ(0, result); + EXPECT_FALSE(adapter->want_write()); + EXPECT_EQ(0, adapter->GetSendWindowSize()); + + auto source = absl::make_unique<TestMetadataSource>(ToHeaderBlock(ToHeaders( + {{"query-cost", "is too darn high"}, {"secret-sauce", "hollandaise"}}))); + adapter->SubmitMetadata(1, 16384u, std::move(source)); + EXPECT_CALL(visitor, OnBeforeFrameSent(kMetadataFrameType, 1, _, 0x4)); + EXPECT_CALL(visitor, OnFrameSent(kMetadataFrameType, 1, _, 0x4, 0)); + + result = adapter->Send(); + EXPECT_EQ(0, result); +} + TEST_F(OgHttp2AdapterTest, SubmitMetadata) { auto source = absl::make_unique<TestMetadataSource>(ToHeaderBlock(ToHeaders( {{"query-cost", "is too darn high"}, {"secret-sauce", "hollandaise"}})));
diff --git a/http2/adapter/oghttp2_session.cc b/http2/adapter/oghttp2_session.cc index 82d9d87..aaf97d7 100644 --- a/http2/adapter/oghttp2_session.cc +++ b/http2/adapter/oghttp2_session.cc
@@ -482,9 +482,8 @@ continue_writing = SendMetadata(0, connection_metadata_); } // Wake streams for writes. - while (continue_writing == SendResult::SEND_OK && - write_scheduler_.HasReadyStreams() && connection_send_window_ > 0) { - const Http2StreamId stream_id = write_scheduler_.PopNextReadyStream(); + while (continue_writing == SendResult::SEND_OK && HasReadyStream()) { + const Http2StreamId stream_id = GetNextReadyStream(); // TODO(birenroy): Add a return value to indicate write blockage, so streams // aren't woken unnecessarily. QUICHE_VLOG(1) << "Waking stream " << stream_id << " for writes."; @@ -496,6 +495,28 @@ return continue_writing == SendResult::SEND_ERROR ? -1 : 0; } +bool OgHttp2Session::HasReadyStream() const { + return !metadata_ready_.empty() || !trailers_ready_.empty() || + (write_scheduler_.HasReadyStreams() && connection_send_window_ > 0); +} + +Http2StreamId OgHttp2Session::GetNextReadyStream() { + QUICHE_DCHECK(HasReadyStream()); + if (!metadata_ready_.empty()) { + const Http2StreamId stream_id = *metadata_ready_.begin(); + // WriteForStream() will re-mark the stream as ready, if necessary. + write_scheduler_.MarkStreamNotReady(stream_id); + return stream_id; + } + if (!trailers_ready_.empty()) { + const Http2StreamId stream_id = *trailers_ready_.begin(); + // WriteForStream() will re-mark the stream as ready, if necessary. + write_scheduler_.MarkStreamNotReady(stream_id); + return stream_id; + } + return write_scheduler_.PopNextReadyStream(); +} + OgHttp2Session::SendResult OgHttp2Session::MaybeSendBufferedData() { int64_t result = std::numeric_limits<int64_t>::max(); while (result > 0 && !buffered_data_.empty()) { @@ -692,7 +713,8 @@ if (connection_can_write != SendResult::SEND_OK) { return connection_can_write; } - return available_window <= 0 ? SendResult::SEND_BLOCKED : SendResult::SEND_OK; + return connection_send_window_ <= 0 ? SendResult::SEND_BLOCKED + : SendResult::SEND_OK; } OgHttp2Session::SendResult OgHttp2Session::SendMetadata( @@ -719,6 +741,7 @@ std::string(payload))); if (end_metadata) { sequence.erase(sequence.begin()); + metadata_ready_.erase(stream_id); } } return SendQueuedFrames(); @@ -789,7 +812,7 @@ state.trailers = absl::make_unique<spdy::SpdyHeaderBlock>(ToHeaderBlock(trailers)); if (!options_.trailers_require_end_data || !iter->second.data_deferred) { - write_scheduler_.MarkStreamReady(stream_id, false); + trailers_ready_.insert(stream_id); } } return 0; @@ -802,7 +825,7 @@ } else { auto iter = CreateStream(stream_id); iter->second.outbound_metadata.push_back(std::move(source)); - write_scheduler_.MarkStreamReady(stream_id, false); + metadata_ready_.insert(stream_id); } } @@ -1089,6 +1112,7 @@ } else { if (it->second.send_window == 0) { // The stream was blocked on flow control. + QUICHE_VLOG(1) << "Marking stream " << stream_id << " ready to write."; write_scheduler_.MarkStreamReady(stream_id, false); } it->second.send_window += delta_window_size; @@ -1277,6 +1301,7 @@ absl::make_unique<spdy::SpdyHeadersIR>(stream_id, std::move(trailers)); frame->set_fin(true); EnqueueFrame(std::move(frame)); + trailers_ready_.erase(stream_id); } void OgHttp2Session::MaybeFinWithRstStream(StreamStateMap::iterator iter) { @@ -1341,6 +1366,8 @@ Http2ErrorCode error_code) { visitor_.OnCloseStream(stream_id, error_code); stream_map_.erase(stream_id); + trailers_ready_.erase(stream_id); + metadata_ready_.erase(stream_id); if (write_scheduler_.StreamRegistered(stream_id)) { write_scheduler_.UnregisterStream(stream_id); }
diff --git a/http2/adapter/oghttp2_session.h b/http2/adapter/oghttp2_session.h index fd7bdc2..e4a2374 100644 --- a/http2/adapter/oghttp2_session.h +++ b/http2/adapter/oghttp2_session.h
@@ -125,7 +125,7 @@ } bool want_write() const override { return !frames_.empty() || !buffered_data_.empty() || - write_scheduler_.HasReadyStreams() || !connection_metadata_.empty(); + !connection_metadata_.empty() || HasReadyStream(); } int GetRemoteWindowSize() const override { return connection_send_window_; } @@ -285,6 +285,13 @@ SEND_ERROR, }; + // Returns true if at least one stream has data or control frames to write. + bool HasReadyStream() const; + + // Returns the next stream that has something to write. If there are no such + // streams, returns zero. + Http2StreamId GetNextReadyStream(); + // Sends the buffered connection preface or serialized frame data, if any. SendResult MaybeSendBufferedData(); @@ -391,7 +398,13 @@ WindowManager connection_window_manager_; absl::flat_hash_set<Http2StreamId> streams_reset_; + + // The number of frames currently queued per stream. absl::flat_hash_map<Http2StreamId, int> queued_frames_; + // Includes streams that are currently ready to write trailers. + absl::flat_hash_set<Http2StreamId> trailers_ready_; + // Includes streams that are currently ready to write metadata. + absl::flat_hash_set<Http2StreamId> metadata_ready_; MetadataSequence connection_metadata_;