Adds a return value to DataFrameSource::Send(), indicating whether the send operation would encounter a write block. PiperOrigin-RevId: 378651262
diff --git a/http2/adapter/data_source.h b/http2/adapter/data_source.h index 5658bfd..2509358 100644 --- a/http2/adapter/data_source.h +++ b/http2/adapter/data_source.h
@@ -23,9 +23,9 @@ virtual std::pair<ssize_t, bool> SelectPayloadLength(size_t max_length) = 0; // This method is called with a frame header and a payload length to send. The - // source should send or buffer the entire frame. - // TODO(birenroy): Consider adding a return value to indicate write blockage. - virtual void Send(absl::string_view frame_header, size_t payload_length) = 0; + // source should send or buffer the entire frame and return true, or return + // false without sending or buffering anything. + virtual bool Send(absl::string_view frame_header, size_t payload_length) = 0; // If true, the end of this data source indicates the end of the stream. // Otherwise, this data will be followed by trailers.
diff --git a/http2/adapter/nghttp2_adapter_test.cc b/http2/adapter/nghttp2_adapter_test.cc index 65a0c7e..d930a26 100644 --- a/http2/adapter/nghttp2_adapter_test.cc +++ b/http2/adapter/nghttp2_adapter_test.cc
@@ -34,7 +34,10 @@ void* user_data) { auto* visitor = static_cast<Http2VisitorInterface*>(user_data); // Send the frame header via the visitor. - visitor->OnReadyToSend(ToStringView(framehd, 9)); + ssize_t result = visitor->OnReadyToSend(ToStringView(framehd, 9)); + if (result == 0) { + return NGHTTP2_ERR_WOULDBLOCK; + } auto* test_source = static_cast<TestDataSource*>(source->ptr); absl::string_view payload = test_source->ReadNext(length); // Send the frame payload via the visitor. @@ -355,6 +358,50 @@ EXPECT_FALSE(adapter->session().want_write()); } +// This test verifies how nghttp2 behaves when a connection becomes +// write-blocked. +TEST(NgHttp2AdapterTest, ClientSubmitRequestWithDataProviderAndWriteBlock) { + 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}; + // 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()); + + visitor.set_is_write_blocked(true); + adapter->Send(); + EXPECT_THAT(visitor.data(), testing::IsEmpty()); + EXPECT_TRUE(adapter->session().want_write()); + + visitor.set_is_write_blocked(false); + 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, + 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_util.cc b/http2/adapter/nghttp2_util.cc index 812bbec..f3743ca 100644 --- a/http2/adapter/nghttp2_util.cc +++ b/http2/adapter/nghttp2_util.cc
@@ -149,10 +149,14 @@ } } - void Send(absl::string_view frame_header, size_t payload_length) override { - send_data_(nullptr /* session */, nullptr /* frame */, - ToUint8Ptr(frame_header.data()), payload_length, - &provider_.source, user_data_); + bool Send(absl::string_view frame_header, size_t payload_length) override { + const int result = + send_data_(nullptr /* session */, nullptr /* frame */, + ToUint8Ptr(frame_header.data()), payload_length, + &provider_.source, user_data_); + QUICHE_LOG_IF(ERROR, result < 0 && result != NGHTTP2_ERR_WOULDBLOCK) + << "Unexpected error code from send: " << result; + return result == 0; } bool send_fin() const override { return send_fin_; }
diff --git a/http2/adapter/oghttp2_session.cc b/http2/adapter/oghttp2_session.cc index 8c10656..bb8405f 100644 --- a/http2/adapter/oghttp2_session.cc +++ b/http2/adapter/oghttp2_session.cc
@@ -138,15 +138,20 @@ // Serialize and send frames in the queue. while (!frames_.empty()) { spdy::SpdySerializedFrame frame = framer_.SerializeFrame(*frames_.front()); - frames_.pop_front(); const ssize_t result = visitor_.OnReadyToSend(absl::string_view(frame)); if (result < 0) { visitor_.OnConnectionError(); return false; - } - if (result < frame.size()) { - serialized_prefix_.assign(frame.data() + result, frame.size() - result); + } else if (result == 0) { + // Write blocked. return false; + } else { + frames_.pop_front(); + if (result < frame.size()) { + // The frame was partially written, so the rest must be buffered. + serialized_prefix_.assign(frame.data() + result, frame.size() - result); + return false; + } } } return true; @@ -174,6 +179,7 @@ return true; } bool source_can_produce = true; + bool connection_can_write = 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) { @@ -193,7 +199,12 @@ data.SetDataShallow(length); spdy::SpdySerializedFrame header = spdy::SpdyFramer::SerializeDataFrameHeaderWithPaddingLengthField(data); - state.outbound_body->Send(absl::string_view(header), length); + const bool success = + state.outbound_body->Send(absl::string_view(header), length); + if (!success) { + connection_can_write = false; + break; + } peer_window_ -= length; state.send_window -= length; available_window = @@ -221,7 +232,9 @@ state.outbound_body != nullptr) { write_scheduler_.MarkStreamReady(stream_id, false); } - return available_window > 0; + // Streams can continue writing as long as the connection is not write-blocked + // and there is additional flow control quota available. + return connection_can_write && available_window > 0; } int32_t OgHttp2Session::SubmitRequest(absl::Span<const Header> headers,
diff --git a/http2/adapter/oghttp2_session_test.cc b/http2/adapter/oghttp2_session_test.cc index 636fda1..4b42188 100644 --- a/http2/adapter/oghttp2_session_test.cc +++ b/http2/adapter/oghttp2_session_test.cc
@@ -256,6 +256,43 @@ // read blocked. } +// This test exercises the case where the connection to the peer is write +// blocked. +TEST(OgHttp2SessionTest, ClientSubmitRequestWithWriteBlock) { + 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."); + 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)); + visitor.set_is_write_blocked(true); + session.Send(); + + EXPECT_THAT(visitor.data(), testing::IsEmpty()); + EXPECT_TRUE(session.want_write()); + visitor.set_is_write_blocked(false); + 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, + SpdyFrameType::DATA})); + EXPECT_FALSE(session.want_write()); +} + TEST(OgHttp2SessionTest, ClientStartShutdown) { DataSavingVisitor visitor; OgHttp2Session session(
diff --git a/http2/adapter/test_utils.cc b/http2/adapter/test_utils.cc index c6e98a2..e0dfec6 100644 --- a/http2/adapter/test_utils.cc +++ b/http2/adapter/test_utils.cc
@@ -48,7 +48,7 @@ return {length, end_data}; } -void TestDataFrameSource::Send(absl::string_view frame_header, +bool TestDataFrameSource::Send(absl::string_view frame_header, size_t payload_length) { QUICHE_LOG_IF(DFATAL, payload_length > current_fragment_.size()) << "payload_length: " << payload_length @@ -56,13 +56,23 @@ const std::string concatenated = absl::StrCat(frame_header, current_fragment_.substr(0, payload_length)); const ssize_t result = visitor_.OnReadyToSend(concatenated); - if (result < concatenated.size()) { - QUICHE_LOG(ERROR) + if (result < 0) { + // Write encountered error. + visitor_.OnConnectionError(); + current_fragment_ = {}; + payload_fragments_.clear(); + return false; + } else if (result == 0) { + // Write blocked. + return false; + } else if (result < concatenated.size()) { + // Probably need to handle this better within this test class. + QUICHE_LOG(DFATAL) << "DATA frame not fully flushed. Connection will be corrupt!"; visitor_.OnConnectionError(); current_fragment_ = {}; payload_fragments_.clear(); - return; + return false; } current_fragment_.remove_prefix(payload_length); if (current_fragment_.empty()) { @@ -71,6 +81,7 @@ current_fragment_ = payload_fragments_.front(); } } + return true; } namespace {
diff --git a/http2/adapter/test_utils.h b/http2/adapter/test_utils.h index de001a1..1d58669 100644 --- a/http2/adapter/test_utils.h +++ b/http2/adapter/test_utils.h
@@ -19,6 +19,9 @@ class DataSavingVisitor : public testing::StrictMock<MockHttp2Visitor> { public: ssize_t OnReadyToSend(absl::string_view data) override { + if (is_write_blocked_) { + return 0; + } const size_t to_accept = std::min(send_limit_, data.size()); absl::StrAppend(&data_, data.substr(0, to_accept)); return to_accept; @@ -29,9 +32,13 @@ void set_send_limit(size_t limit) { send_limit_ = limit; } + bool is_write_blocked() const { return is_write_blocked_; } + void set_is_write_blocked(bool value) { is_write_blocked_ = value; } + private: std::string data_; size_t send_limit_ = std::numeric_limits<size_t>::max(); + bool is_write_blocked_ = false; }; // A test DataFrameSource that can be initialized with a single string payload, @@ -47,7 +54,7 @@ bool has_fin = true); std::pair<ssize_t, bool> SelectPayloadLength(size_t max_length) override; - void Send(absl::string_view frame_header, size_t payload_length) override; + bool 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; }