wwbmmm commented on code in PR #3690:
URL: https://github.com/apache/brpc/pull/3690#discussion_r4500539185
##########
src/brpc/policy/http2_rpc_protocol.cpp:
##########
@@ -2019,57 +2182,151 @@ H2UnsentResponse::AppendAndDestroySelf(butil::IOBuf*
out, Socket* socket) {
return butil::Status::OK();
}
H2Context* ctx = static_cast<H2Context*>(socket->parsing_context());
+ if (ctx == NULL || ctx->is_client_side() || _stream_id == 0) {
+ return butil::Status::OK();
+ }
+
+ H2StreamContext* overcrowded_sctx = nullptr;
+ H2StreamContext* finished_sctx = nullptr;
+ H2StreamContext* notify_failed_sctx = nullptr;
+ bthread_id_t deferred_notify_id = INVALID_BTHREAD_ID;
+ int notify_reg_error = 0;
+ {
+ std::unique_lock<butil::Mutex> mu(ctx->_stream_mutex);
+ H2StreamContext** psctx = ctx->_pending_streams.seek(_stream_id);
+ if (psctx == nullptr || (*psctx)->_rejected_error != H2_NO_ERROR) {
+ // Stream was already reset by client, abandoned, or rejected.
+ return butil::Status::OK();
+ }
+ H2StreamContext* sctx = *psctx;
+ const int64_t conn_window =
+ ctx->_remote_window_left.load(butil::memory_order_relaxed);
+ const int64_t stream_window =
+ sctx->_remote_window_left.load(butil::memory_order_relaxed);
+ size_t sendable_size = 0;
+ if (conn_window > 0 && stream_window > 0) {
+ sendable_size = std::min(
+ _data.size(),
+ static_cast<size_t>(std::min(conn_window, stream_window)));
+ }
+ const size_t pending_size = _data.size() - sendable_size;
+ const int64_t max_pending = GetH2MaxPendingDataBytes();
+ if (max_pending > 0 && pending_size > 0) {
+ const auto limit = static_cast<size_t>(max_pending);
+ const size_t cur_pending =
+ ctx->_pending_data_size.load(butil::memory_order_relaxed);
+ if (cur_pending > limit || pending_size > limit - cur_pending) {
+ overcrowded_sctx = ctx->RemoveStreamLocked(_stream_id);
+ }
+ }
+ if (overcrowded_sctx == nullptr) {
+ HPacker& hpacker = ctx->hpacker();
+ butil::IOBufAppender appender;
+ HPackOptions options;
+ options.encode_name = FLAGS_h2_hpack_encode_name;
+ options.encode_value = FLAGS_h2_hpack_encode_value;
+ if (ctx->_remote_settings.header_table_size == 0) {
+ options.index_policy = HPACK_NEVER_INDEX_HEADER;
+ }
+ for (size_t i = 0; i < _size; ++i) {
+ hpacker.Encode(&appender, _list[i], options);
+ }
+ if (_http_response) {
+ for (HttpHeader::HeaderIterator it =
_http_response->HeaderBegin();
+ it != _http_response->HeaderEnd(); ++it) {
+ HPacker::Header header(it->first, it->second);
+ hpacker.Encode(&appender, header, options);
+ }
+ }
+ butil::IOBuf frag;
+ appender.move_to(frag);
+ PackH2Headers(out, frag, _stream_id,
+ ctx->_remote_settings.max_frame_size,
+ _data.empty() && !_is_grpc);
+
+ if (_is_grpc) {
+ HPackOptions trailer_options = options;
+ trailer_options.index_policy = HPACK_NOT_INDEX_HEADER;
+ HPacker::Header status_header(
+ "grpc-status", butil::string_printf("%d", _grpc_status));
+ HPacker::EncodeWithoutDynamicTable(
+ &appender, status_header, trailer_options);
+ if (!_grpc_message.empty()) {
+ HPacker::Header msg_header("grpc-message", _grpc_message);
+ HPacker::EncodeWithoutDynamicTable(
+ &appender, msg_header, trailer_options);
+ }
+ appender.move_to(sctx->_pending_trailer);
+ }
- // flow control
- // NOTE: Currently the stream context is definitely removed and updating
- // window size is useless, however it's not true when progressive request
- // is supported.
- // TODO(zhujiashun): Instead of just returning error to client, a better
- // solution to handle not enough window size is to wait until WINDOW_UPDATE
- // is received, and then retry those failed response again.
- if (!MinusWindowSize(&ctx->_remote_window_left, _data.size())) {
+ sctx->_response_committed = true;
+ if (!_data.empty()) {
+ CHECK(sctx->_pending_data.empty());
+ sctx->_pending_data = _data;
+ ctx->AddPendingDataSize(_data.size());
+ }
+ ctx->AppendPendingDataLocked(sctx, out);
+ if (sctx->_pending_data.empty() && sctx->_pending_trailer.empty())
{
+ finished_sctx = ctx->RemoveStreamLocked(_stream_id);
Review Comment:
[replied by brpc-oncall robot] Erasing the stream here (and the two
`RemoveStreamLocked()` calls in `FlushPendingData()`) drops the stream's
deferred request-side WINDOW_UPDATE, unlike `RemoveStreamAndDeferWU()` (also
used by `OnSocketFailed`) and the overcrowded / notify-failure branches above,
which all call `DeferWindowUpdate(sctx->ReleaseDeferredWindowUpdate())`. This
matters when the response completes before the request does (a progressive-read
method whose handler answers early, i.e. `_remote_ended == false`): the request
bytes already buffered in `H2StreamContext::_deferred_window_update` (up to one
per-stream window quota) are never added back to the connection window, the
peer never regains that connection-level send credit, and after enough early
responses the peer's connection window is exhausted and the connection stalls
even though every remaining stream looks idle. Please migrate the credit on
every server-side erase of a response stream (collect it under `_stream_m
utex` and call `DeferWindowUpdate()` after unlocking, as the overcrowded path
does).
In the same early-response scenario the remaining request DATA frames hit
the `FindOpenStream() == nullptr` branch of `H2Context::OnData()` and log
`LOG(ERROR) "Fail to find stream_id"` per frame; that is a normal flow now (RFC
9113 section 8.1 would have us send `RST_STREAM(NO_ERROR)` after a complete
response), so the log level should be relaxed there.
##########
test/brpc_h2_unsent_message_unittest.cpp:
##########
@@ -216,7 +221,7 @@ TEST(H2UnsentMessage,
request_does_not_fail_when_body_exceeds_window) {
brpc::policy::H2StreamContext* sctx = ctx->FindStream(1);
ASSERT_TRUE(sctx != nullptr);
EXPECT_EQ(7u, sctx->_pending_data.size());
- EXPECT_EQ(7u, ctx->_pending_data_size);
+ EXPECT_EQ(7u, ctx->_pending_data_size.load());
EXPECT_EQ(0, ctx->_remote_window_left);
EXPECT_EQ(0, sctx->_remote_window_left);
Review Comment:
[replied by brpc-oncall robot] `sctx->Destroy()` releases the reference that
`_pending_streams` owns (`TryToInsertStream()` took over the constructor
reference), while the stream is still in the map. It happens to be harmless
only because this test never destroys the `H2Context`; on a teardown path that
does, `~H2Context()` would call `RemoveOneRefForStage2()` on the already-freed
stream (double release / use-after-free). Please use `RemoveStreamAndDeferWU()`
+ `RemoveOneRefForStage2()` here, or let the context release the stream.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
