Implements read block detection in OgHttp2Session, and fixes read block handling in Nghttp2DataFrameSource. Note that currently, there is no way to reschedule a stream that encounters a read block in the OgHttp2 stack. (Coming in a future change.) PiperOrigin-RevId: 378512810
diff --git a/http2/adapter/nghttp2_adapter_test.cc b/http2/adapter/nghttp2_adapter_test.cc index ded57cc..65a0c7e 100644 --- a/http2/adapter/nghttp2_adapter_test.cc +++ b/http2/adapter/nghttp2_adapter_test.cc
@@ -24,6 +24,24 @@ WINDOW_UPDATE, }; +// This send callback assumes |source|'s pointer is a TestDataSource, and +// |user_data| is a Http2VisitorInterface. +int TestSendCallback(nghttp2_session*, + nghttp2_frame* frame, + const uint8_t* framehd, + size_t length, + nghttp2_data_source* source, + void* user_data) { + auto* visitor = static_cast<Http2VisitorInterface*>(user_data); + // Send the frame header via the visitor. + visitor->OnReadyToSend(ToStringView(framehd, 9)); + auto* test_source = static_cast<TestDataSource*>(source->ptr); + absl::string_view payload = test_source->ReadNext(length); + // Send the frame payload via the visitor. + visitor->OnReadyToSend(payload); + return 0; +} + TEST(NgHttp2AdapterTest, ClientConstruction) { testing::StrictMock<MockHttp2Visitor> visitor; auto adapter = NgHttp2Adapter::CreateClientAdapter(visitor); @@ -269,22 +287,8 @@ TestDataSource body1{kBody}; // The TestDataSource is wrapped in the nghttp2_data_provider data type. nghttp2_data_provider provider = body1.MakeDataProvider(); + nghttp2_send_data_callback send_callback = &TestSendCallback; - // This send callback assumes |source|'s pointer is a TestDataSource, which we - // know is true because we just converted |body1| into a nghttp2_data_provider - // above. - nghttp2_send_data_callback send_callback = - [](nghttp2_session*, nghttp2_frame* frame, const uint8_t* framehd, - size_t length, nghttp2_data_source* source, void* user_data) { - auto* visitor = static_cast<Http2VisitorInterface*>(user_data); - // Send the frame header via the visitor. - visitor->OnReadyToSend(ToStringView(framehd, 9)); - auto* test_source = static_cast<TestDataSource*>(source->ptr); - absl::string_view payload = test_source->ReadNext(length); - // Send the frame payload via the visitor. - visitor->OnReadyToSend(payload); - return 0; - }; // This call transforms it back into a DataFrameSource, which is compatible // with the Http2Adapter API. std::unique_ptr<DataFrameSource> frame_source = @@ -304,6 +308,53 @@ EXPECT_FALSE(adapter->session().want_write()); } +// This test verifies how nghttp2 behaves when a data source becomes +// read-blocked. +TEST(NgHttp2AdapterTest, ClientSubmitRequestWithDataProviderAndReadBlock) { + DataSavingVisitor visitor; + auto adapter = NgHttp2Adapter::CreateClientAdapter(visitor); + + const absl::string_view kBody = "This is an example request body."; + // This test will use TestDataSource as the source of the body payload data. + TestDataSource body1{kBody}; + body1.set_is_data_available(false); + // The TestDataSource is wrapped in the nghttp2_data_provider data type. + nghttp2_data_provider provider = body1.MakeDataProvider(); + nghttp2_send_data_callback send_callback = &TestSendCallback; + + // This call transforms it back into a DataFrameSource, which is compatible + // with the Http2Adapter API. + std::unique_ptr<DataFrameSource> frame_source = + MakeZeroCopyDataFrameSource(provider, &visitor, std::move(send_callback)); + int stream_id = + adapter->SubmitRequest(ToHeaders({{":method", "POST"}, + {":scheme", "http"}, + {":authority", "example.com"}, + {":path", "/this/is/request/one"}}), + frame_source.get(), nullptr); + EXPECT_GT(stream_id, 0); + EXPECT_TRUE(adapter->session().want_write()); + + adapter->Send(); + // Client preface does not appear to include the mandatory SETTINGS frame. + absl::string_view serialized = visitor.data(); + EXPECT_THAT(serialized, + testing::StartsWith(spdy::kHttp2ConnectionHeaderPrefix)); + serialized.remove_prefix(strlen(spdy::kHttp2ConnectionHeaderPrefix)); + EXPECT_THAT(serialized, EqualsFrames({spdy::SpdyFrameType::HEADERS})); + visitor.Clear(); + EXPECT_FALSE(adapter->session().want_write()); + + // Resume the deferred stream. + body1.set_is_data_available(true); + nghttp2_session_resume_data(adapter->session().raw_ptr(), stream_id); + EXPECT_TRUE(adapter->session().want_write()); + + adapter->Send(); + EXPECT_THAT(visitor.data(), EqualsFrames({spdy::SpdyFrameType::DATA})); + EXPECT_FALSE(adapter->session().want_write()); +} + TEST(NgHttp2AdapterTest, ServerConstruction) { testing::StrictMock<MockHttp2Visitor> visitor; auto adapter = NgHttp2Adapter::CreateServerAdapter(visitor);
diff --git a/http2/adapter/nghttp2_data_provider.cc b/http2/adapter/nghttp2_data_provider.cc index 7a70965..83037e4 100644 --- a/http2/adapter/nghttp2_data_provider.cc +++ b/http2/adapter/nghttp2_data_provider.cc
@@ -18,6 +18,7 @@ uint32_t* data_flags, nghttp2_data_source* source, void* /* user_data */) { + *data_flags |= NGHTTP2_DATA_FLAG_NO_COPY; auto* frame_source = static_cast<DataFrameSource*>(source->ptr); auto [result_length, done] = frame_source->SelectPayloadLength(length); if (result_length == DataFrameSource::kBlocked) { @@ -25,7 +26,6 @@ } else if (result_length == DataFrameSource::kError) { return NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE; } - *data_flags |= NGHTTP2_DATA_FLAG_NO_COPY; if (done) { *data_flags |= NGHTTP2_DATA_FLAG_EOF; }
diff --git a/http2/adapter/nghttp2_util.cc b/http2/adapter/nghttp2_util.cc index d97c64a..812bbec 100644 --- a/http2/adapter/nghttp2_util.cc +++ b/http2/adapter/nghttp2_util.cc
@@ -133,11 +133,13 @@ ssize_t result = provider_.read_callback( nullptr /* session */, stream_id, nullptr /* buf */, max_length, &data_flags, &provider_.source, nullptr /* user_data */); - if (result < 0) { - return {-1, false}; + if (result == NGHTTP2_ERR_DEFERRED) { + return {kBlocked, false}; + } else if (result < 0) { + return {kError, false}; } else if ((data_flags & NGHTTP2_DATA_FLAG_NO_COPY) == 0) { QUICHE_LOG(ERROR) << "Source did not use the zero-copy API!"; - return {-1, false}; + return {kError, false}; } else { if (data_flags & NGHTTP2_DATA_FLAG_NO_END_STREAM) { send_fin_ = false;
diff --git a/http2/adapter/oghttp2_session.cc b/http2/adapter/oghttp2_session.cc index 330476c..8c10656 100644 --- a/http2/adapter/oghttp2_session.cc +++ b/http2/adapter/oghttp2_session.cc
@@ -127,7 +127,7 @@ const Http2StreamId stream_id = write_scheduler_.PopNextReadyStream(); // TODO(birenroy): Add a return value to indicate write blockage, so streams // aren't woken unnecessarily. - WriteForStream(stream_id); + continue_writing = WriteForStream(stream_id); } if (continue_writing) { SendQueuedFrames(); @@ -152,12 +152,12 @@ return true; } -void OgHttp2Session::WriteForStream(Http2StreamId stream_id) { +bool OgHttp2Session::WriteForStream(Http2StreamId stream_id) { auto it = stream_map_.find(stream_id); if (it == stream_map_.end()) { QUICHE_LOG(ERROR) << "Can't find stream " << stream_id << " which is ready to write!"; - return; + return true; } StreamState& state = it->second; if (state.outbound_body == nullptr) { @@ -171,13 +171,22 @@ MaybeCloseWithRstStream(stream_id, state); } } - return; + return true; } + bool source_can_produce = true; int32_t available_window = std::min(std::min(peer_window_, state.send_window), max_frame_payload_); while (available_window > 0 && state.outbound_body != nullptr) { auto [length, end_data] = state.outbound_body->SelectPayloadLength(available_window); + if (length == DataFrameSource::kBlocked) { + source_can_produce = false; + break; + } else if (length == DataFrameSource::kError) { + source_can_produce = false; + visitor_.OnCloseStream(stream_id, Http2ErrorCode::INTERNAL_ERROR); + break; + } const bool fin = end_data ? state.outbound_body->send_fin() : false; spdy::SpdyDataIR data(stream_id); data.set_fin(fin); @@ -208,9 +217,11 @@ } // If the stream still has data to send, it should be marked as ready in the // write scheduler. - if (state.send_window > 0 && state.outbound_body != nullptr) { + if (source_can_produce && state.send_window > 0 && + state.outbound_body != nullptr) { write_scheduler_.MarkStreamReady(stream_id, false); } + return available_window > 0; } int32_t OgHttp2Session::SubmitRequest(absl::Span<const Header> headers,
diff --git a/http2/adapter/oghttp2_session.h b/http2/adapter/oghttp2_session.h index 30bc9d8..c5b9efc 100644 --- a/http2/adapter/oghttp2_session.h +++ b/http2/adapter/oghttp2_session.h
@@ -160,7 +160,10 @@ // Sends queued frames, returning true if all frames were flushed. bool SendQueuedFrames(); - void WriteForStream(Http2StreamId stream_id); + // Returns false if the connection is write-blocked (due to flow control or + // some other reason). + bool WriteForStream(Http2StreamId stream_id); + void SendTrailers(Http2StreamId stream_id, spdy::SpdyHeaderBlock trailers); // Encapsulates the RST_STREAM NO_ERROR behavior described in RFC 7540
diff --git a/http2/adapter/oghttp2_session_test.cc b/http2/adapter/oghttp2_session_test.cc index 185fd3a..636fda1 100644 --- a/http2/adapter/oghttp2_session_test.cc +++ b/http2/adapter/oghttp2_session_test.cc
@@ -221,6 +221,41 @@ EXPECT_THAT(visitor.data(), EqualsFrames({spdy::SpdyFrameType::HEADERS})); } +// This test exercises the case where the client request body source is read +// blocked. +TEST(OgHttp2SessionTest, ClientSubmitRequestWithReadBlock) { + DataSavingVisitor visitor; + OgHttp2Session session( + visitor, OgHttp2Session::Options{.perspective = Perspective::kClient}); + EXPECT_FALSE(session.want_write()); + + const char* kSentinel1 = "arbitrary pointer 1"; + TestDataFrameSource body1(visitor, "This is an example request body."); + body1.set_is_data_available(false); + int stream_id = + session.SubmitRequest(ToHeaders({{":method", "POST"}, + {":scheme", "http"}, + {":authority", "example.com"}, + {":path", "/this/is/request/one"}}), + &body1, const_cast<char*>(kSentinel1)); + EXPECT_GT(stream_id, 0); + EXPECT_TRUE(session.want_write()); + EXPECT_EQ(kSentinel1, session.GetStreamUserData(stream_id)); + session.Send(); + absl::string_view serialized = visitor.data(); + EXPECT_THAT(serialized, + testing::StartsWith(spdy::kHttp2ConnectionHeaderPrefix)); + serialized.remove_prefix(strlen(spdy::kHttp2ConnectionHeaderPrefix)); + EXPECT_THAT(serialized, + EqualsFrames({SpdyFrameType::SETTINGS, SpdyFrameType::HEADERS})); + // No data frame, as body1 was read blocked. + visitor.Clear(); + EXPECT_FALSE(session.want_write()); + + // Currently there is no way to indicate that the first stream is no longer + // read blocked. +} + TEST(OgHttp2SessionTest, ClientStartShutdown) { DataSavingVisitor visitor; OgHttp2Session session(
diff --git a/http2/adapter/test_utils.cc b/http2/adapter/test_utils.cc index 3ccd63e..c6e98a2 100644 --- a/http2/adapter/test_utils.cc +++ b/http2/adapter/test_utils.cc
@@ -36,13 +36,16 @@ std::pair<ssize_t, bool> TestDataFrameSource::SelectPayloadLength( size_t max_length) { + if (!is_data_available_) { + return {kBlocked, false}; + } // The stream is done if there's no more data, or if |max_length| is at least // as large as the remaining data. const bool end_data = current_fragment_.empty() || (payload_fragments_.size() == 1 && max_length >= current_fragment_.size()); const ssize_t length = std::min(max_length, current_fragment_.size()); - return std::make_pair(length, end_data); + return {length, end_data}; } void TestDataFrameSource::Send(absl::string_view frame_header,
diff --git a/http2/adapter/test_utils.h b/http2/adapter/test_utils.h index 0ea7ec4..de001a1 100644 --- a/http2/adapter/test_utils.h +++ b/http2/adapter/test_utils.h
@@ -50,11 +50,14 @@ void Send(absl::string_view frame_header, size_t payload_length) override; bool send_fin() const override { return has_fin_; } + void set_is_data_available(bool value) { is_data_available_ = value; } + private: Http2VisitorInterface& visitor_; std::vector<std::string> payload_fragments_; absl::string_view current_fragment_; const bool has_fin_; + bool is_data_available_ = true; }; // A simple class that can easily be adapted to act as a nghttp2_data_source. @@ -78,20 +81,27 @@ .source = {.ptr = this}, .read_callback = [](nghttp2_session*, int32_t, uint8_t*, size_t length, uint32_t* data_flags, nghttp2_data_source* source, - void*) { + void*) -> ssize_t { + *data_flags |= NGHTTP2_DATA_FLAG_NO_COPY; auto* s = static_cast<TestDataSource*>(source->ptr); + if (!s->is_data_available()) { + return NGHTTP2_ERR_DEFERRED; + } const ssize_t ret = s->SelectPayloadLength(length); if (ret < length) { *data_flags |= NGHTTP2_DATA_FLAG_EOF; } - *data_flags |= NGHTTP2_DATA_FLAG_NO_COPY; return ret; }}; } + bool is_data_available() const { return is_data_available_; } + void set_is_data_available(bool value) { is_data_available_ = value; } + private: const std::string data_; absl::string_view remaining_ = data_; + bool is_data_available_ = true; }; // These matchers check whether a string consists entirely of HTTP/2 frames of