Adds a StreamState member that tracks whether the data source is currently blocked. This affects whether the stream should be scheduled for writes, and therefore the return value of OgHttp2Adapter::want_write(). PiperOrigin-RevId: 401817911
diff --git a/http2/adapter/nghttp2_adapter_test.cc b/http2/adapter/nghttp2_adapter_test.cc index ccfb1ac..62f3682 100644 --- a/http2/adapter/nghttp2_adapter_test.cc +++ b/http2/adapter/nghttp2_adapter_test.cc
@@ -1520,6 +1520,117 @@ spdy::SpdyFrameType::PING})); } +// Tests the case where the response body is in the progress of being sent while +// trailers are queued. +TEST(NgHttp2AdapterTest, ServerSubmitsTrailersWhileDataDeferred) { + DataSavingVisitor visitor; + auto adapter = NgHttp2Adapter::CreateServerAdapter(visitor); + + const std::string frames = TestFrameSequence() + .ClientPreface() + .Headers(1, + {{":method", "POST"}, + {":scheme", "https"}, + {":authority", "example.com"}, + {":path", "/this/is/request/one"}}, + /*fin=*/false) + .WindowUpdate(1, 2000) + .Data(1, "This is the request body.") + .WindowUpdate(0, 2000) + .Serialize(); + testing::InSequence s; + + // Client preface (empty SETTINGS) + EXPECT_CALL(visitor, OnFrameHeader(0, 0, SETTINGS, 0)); + EXPECT_CALL(visitor, OnSettingsStart()); + EXPECT_CALL(visitor, OnSettingsEnd()); + + EXPECT_CALL(visitor, OnFrameHeader(1, _, HEADERS, 4)); + EXPECT_CALL(visitor, OnBeginHeadersForStream(1)); + EXPECT_CALL(visitor, OnHeaderForStream(1, ":method", "POST")); + EXPECT_CALL(visitor, OnHeaderForStream(1, ":scheme", "https")); + EXPECT_CALL(visitor, OnHeaderForStream(1, ":authority", "example.com")); + EXPECT_CALL(visitor, OnHeaderForStream(1, ":path", "/this/is/request/one")); + EXPECT_CALL(visitor, OnEndHeadersForStream(1)); + EXPECT_CALL(visitor, OnFrameHeader(1, 4, WINDOW_UPDATE, 0)); + EXPECT_CALL(visitor, OnWindowUpdate(1, 2000)); + EXPECT_CALL(visitor, OnFrameHeader(1, _, DATA, 0)); + EXPECT_CALL(visitor, OnBeginDataForStream(1, _)); + EXPECT_CALL(visitor, OnDataForStream(1, "This is the request body.")); + EXPECT_CALL(visitor, OnFrameHeader(0, 4, WINDOW_UPDATE, 0)); + EXPECT_CALL(visitor, OnWindowUpdate(0, 2000)); + + const int64_t result = adapter->ProcessBytes(frames); + EXPECT_EQ(frames.size(), static_cast<size_t>(result)); + + EXPECT_TRUE(adapter->want_write()); + + EXPECT_CALL(visitor, OnBeforeFrameSent(SETTINGS, 0, 0, 0x1)); + EXPECT_CALL(visitor, OnFrameSent(SETTINGS, 0, 0, 0x1, 0)); + + int send_result = adapter->Send(); + // Some bytes should have been serialized. + EXPECT_EQ(0, send_result); + // SETTINGS ack + EXPECT_THAT(visitor.data(), EqualsFrames({spdy::SpdyFrameType::SETTINGS})); + visitor.Clear(); + + const absl::string_view kBody = "This is an example response body."; + + // The body source must indicate that the end of the body is not the end of + // the stream. + auto body1 = absl::make_unique<TestDataFrameSource>(visitor, false); + body1->AppendPayload(kBody); + auto* body1_ptr = body1.get(); + int submit_result = adapter->SubmitResponse( + 1, ToHeaders({{":status", "200"}, {"x-comment", "Sure, sounds good."}}), + std::move(body1)); + EXPECT_EQ(submit_result, 0); + EXPECT_TRUE(adapter->want_write()); + + EXPECT_CALL(visitor, OnBeforeFrameSent(HEADERS, 1, _, 0x4)); + EXPECT_CALL(visitor, OnFrameSent(HEADERS, 1, _, 0x4, 0)); + EXPECT_CALL(visitor, OnFrameSent(DATA, 1, _, 0x0, 0)); + + send_result = adapter->Send(); + // Some bytes should have been serialized. + EXPECT_EQ(0, send_result); + // SETTINGS ack + EXPECT_THAT(visitor.data(), EqualsFrames({spdy::SpdyFrameType::HEADERS, + spdy::SpdyFrameType::DATA})); + visitor.Clear(); + EXPECT_FALSE(adapter->want_write()); + + int trailer_result = + adapter->SubmitTrailer(1, ToHeaders({{"final-status", "a-ok"}})); + ASSERT_EQ(trailer_result, 0); + + // Even though the data source has not finished sending data, nghttp2 will + // write the trailers anyway. + EXPECT_TRUE(adapter->want_write()); + + EXPECT_CALL(visitor, OnBeforeFrameSent(HEADERS, 1, _, 0x5)); + EXPECT_CALL(visitor, OnFrameSent(HEADERS, 1, _, 0x5, 0)); + + send_result = adapter->Send(); + EXPECT_EQ(0, send_result); + EXPECT_THAT(visitor.data(), EqualsFrames({spdy::SpdyFrameType::HEADERS})); + visitor.Clear(); + + // Resuming the stream results in the library wanting to write again. + body1_ptr->AppendPayload(kBody); + body1_ptr->EndData(); + adapter->ResumeStream(1); + EXPECT_TRUE(adapter->want_write()); + + send_result = adapter->Send(); + EXPECT_EQ(0, send_result); + + // But no data is written for the stream. + EXPECT_THAT(visitor.data(), testing::IsEmpty()); + EXPECT_FALSE(adapter->want_write()); +} + TEST(NgHttp2AdapterTest, ServerErrorWhileHandlingHeaders) { DataSavingVisitor visitor; auto adapter = NgHttp2Adapter::CreateServerAdapter(visitor);
diff --git a/http2/adapter/oghttp2_adapter_test.cc b/http2/adapter/oghttp2_adapter_test.cc index 413db77..4731e48 100644 --- a/http2/adapter/oghttp2_adapter_test.cc +++ b/http2/adapter/oghttp2_adapter_test.cc
@@ -1165,6 +1165,112 @@ EXPECT_THAT(visitor.data(), EqualsFrames({spdy::SpdyFrameType::HEADERS})); } +// Tests the case where the response body is in the progress of being sent while +// trailers are queued. +TEST(OgHttp2AdapterServerTest, ServerSubmitsTrailersWhileDataDeferred) { + DataSavingVisitor visitor; + OgHttp2Adapter::Options options{.perspective = Perspective::kServer}; + auto adapter = OgHttp2Adapter::Create(visitor, options); + + const std::string frames = TestFrameSequence() + .ClientPreface() + .Headers(1, + {{":method", "POST"}, + {":scheme", "https"}, + {":authority", "example.com"}, + {":path", "/this/is/request/one"}}, + /*fin=*/false) + .WindowUpdate(1, 2000) + .Data(1, "This is the request body.") + .WindowUpdate(0, 2000) + .Serialize(); + testing::InSequence s; + + // Client preface (empty SETTINGS) + EXPECT_CALL(visitor, OnFrameHeader(0, 0, SETTINGS, 0)); + EXPECT_CALL(visitor, OnSettingsStart()); + EXPECT_CALL(visitor, OnSettingsEnd()); + + EXPECT_CALL(visitor, OnFrameHeader(1, _, HEADERS, 4)); + EXPECT_CALL(visitor, OnBeginHeadersForStream(1)); + EXPECT_CALL(visitor, OnHeaderForStream(1, ":method", "POST")); + EXPECT_CALL(visitor, OnHeaderForStream(1, ":scheme", "https")); + EXPECT_CALL(visitor, OnHeaderForStream(1, ":authority", "example.com")); + EXPECT_CALL(visitor, OnHeaderForStream(1, ":path", "/this/is/request/one")); + EXPECT_CALL(visitor, OnEndHeadersForStream(1)); + EXPECT_CALL(visitor, OnFrameHeader(1, 4, WINDOW_UPDATE, 0)); + EXPECT_CALL(visitor, OnWindowUpdate(1, 2000)); + EXPECT_CALL(visitor, OnFrameHeader(1, _, DATA, 0)); + EXPECT_CALL(visitor, OnBeginDataForStream(1, _)); + EXPECT_CALL(visitor, OnDataForStream(1, "This is the request body.")); + EXPECT_CALL(visitor, OnFrameHeader(0, 4, WINDOW_UPDATE, 0)); + EXPECT_CALL(visitor, OnWindowUpdate(0, 2000)); + + const int64_t result = adapter->ProcessBytes(frames); + EXPECT_EQ(frames.size(), static_cast<size_t>(result)); + + EXPECT_TRUE(adapter->want_write()); + + EXPECT_CALL(visitor, OnBeforeFrameSent(SETTINGS, 0, _, 0x0)); + EXPECT_CALL(visitor, OnFrameSent(SETTINGS, 0, _, 0x0, 0)); + EXPECT_CALL(visitor, OnBeforeFrameSent(SETTINGS, 0, 0, 0x1)); + EXPECT_CALL(visitor, OnFrameSent(SETTINGS, 0, 0, 0x1, 0)); + + int send_result = adapter->Send(); + // Some bytes should have been serialized. + EXPECT_EQ(0, send_result); + // SETTINGS ack + EXPECT_THAT(visitor.data(), EqualsFrames({spdy::SpdyFrameType::SETTINGS, + spdy::SpdyFrameType::SETTINGS})); + visitor.Clear(); + + const absl::string_view kBody = "This is an example response body."; + + // The body source must indicate that the end of the body is not the end of + // the stream. + auto body1 = absl::make_unique<TestDataFrameSource>(visitor, false); + body1->AppendPayload(kBody); + auto* body1_ptr = body1.get(); + int submit_result = adapter->SubmitResponse( + 1, ToHeaders({{":status", "200"}, {"x-comment", "Sure, sounds good."}}), + std::move(body1)); + EXPECT_EQ(submit_result, 0); + EXPECT_TRUE(adapter->want_write()); + + EXPECT_CALL(visitor, OnBeforeFrameSent(HEADERS, 1, _, 0x4)); + EXPECT_CALL(visitor, OnFrameSent(HEADERS, 1, _, 0x4, 0)); + EXPECT_CALL(visitor, OnFrameSent(DATA, 1, _, 0x0, 0)); + + send_result = adapter->Send(); + // Some bytes should have been serialized. + EXPECT_EQ(0, send_result); + // SETTINGS ack + EXPECT_THAT(visitor.data(), EqualsFrames({spdy::SpdyFrameType::HEADERS, + spdy::SpdyFrameType::DATA})); + visitor.Clear(); + EXPECT_FALSE(adapter->want_write()); + + int trailer_result = + adapter->SubmitTrailer(1, ToHeaders({{"final-status", "a-ok"}})); + ASSERT_EQ(trailer_result, 0); + // Even though there are new trailers to write, the data source has not + // finished writing data and is blocked. + EXPECT_FALSE(adapter->want_write()); + + body1_ptr->EndData(); + adapter->ResumeStream(1); + EXPECT_TRUE(adapter->want_write()); + + EXPECT_CALL(visitor, OnBeforeFrameSent(HEADERS, 1, _, 0x5)); + EXPECT_CALL(visitor, OnFrameSent(HEADERS, 1, _, 0x5, 0)); + EXPECT_CALL(visitor, OnBeforeFrameSent(RST_STREAM, 1, _, 0x0)); + EXPECT_CALL(visitor, OnFrameSent(RST_STREAM, 1, _, 0x0, 0)); + EXPECT_CALL(visitor, OnCloseStream(1, Http2ErrorCode::NO_ERROR)); + + send_result = adapter->Send(); + EXPECT_EQ(0, send_result); +} + TEST(OgHttp2AdapterServerTest, ServerErrorWhileHandlingHeaders) { DataSavingVisitor visitor; OgHttp2Adapter::Options options{.perspective = Perspective::kServer};
diff --git a/http2/adapter/oghttp2_session.cc b/http2/adapter/oghttp2_session.cc index e3a696b..09a3ce0 100644 --- a/http2/adapter/oghttp2_session.cc +++ b/http2/adapter/oghttp2_session.cc
@@ -216,10 +216,11 @@ bool OgHttp2Session::ResumeStream(Http2StreamId stream_id) { auto it = stream_map_.find(stream_id); - if (it->second.outbound_body == nullptr || + if (it == stream_map_.end() || it->second.outbound_body == nullptr || !write_scheduler_.StreamRegistered(stream_id)) { return false; } + it->second.data_deferred = false; write_scheduler_.MarkStreamReady(stream_id, /*add_to_front=*/false); return true; } @@ -432,7 +433,6 @@ } return true; } - bool source_can_produce = true; int32_t available_window = std::min({connection_send_window_, state.send_window, static_cast<int32_t>(max_frame_payload_)}); @@ -443,10 +443,9 @@ std::tie(length, end_data) = state.outbound_body->SelectPayloadLength(available_window); if (length == 0 && !end_data) { - source_can_produce = false; + state.data_deferred = true; break; } else if (length == DataFrameSource::kError) { - source_can_produce = false; CloseStream(stream_id, Http2ErrorCode::INTERNAL_ERROR); // No more work on the stream; it has been closed. break; @@ -494,7 +493,7 @@ } // If the stream still exists and has data to send, it should be marked as // ready in the write scheduler. - if (stream_map_.contains(stream_id) && source_can_produce && + if (stream_map_.contains(stream_id) && !state.data_deferred && state.send_window > 0 && state.outbound_body != nullptr) { write_scheduler_.MarkStreamReady(stream_id, false); } @@ -601,7 +600,9 @@ // Save trailers so they can be written once data is done. state.trailers = absl::make_unique<spdy::SpdyHeaderBlock>(ToHeaderBlock(trailers)); - write_scheduler_.MarkStreamReady(stream_id, false); + if (!iter->second.data_deferred) { + write_scheduler_.MarkStreamReady(stream_id, false); + } } return 0; }
diff --git a/http2/adapter/oghttp2_session.h b/http2/adapter/oghttp2_session.h index 75f9c96..814b843 100644 --- a/http2/adapter/oghttp2_session.h +++ b/http2/adapter/oghttp2_session.h
@@ -185,6 +185,8 @@ absl::optional<HeaderType> received_header_type; bool half_closed_local = false; bool half_closed_remote = false; + // Indicates that `outbound_body` temporarily cannot produce data. + bool data_deferred = false; }; using StreamStateMap = absl::flat_hash_map<Http2StreamId, StreamState>;