hjwsm1989 commented on code in PR #3690:
URL: https://github.com/apache/brpc/pull/3690#discussion_r4502361412
##########
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:
Thanks for catching this! Fixed in `6605a1a1`:
- Migrated `finished_sctx->ReleaseDeferredWindowUpdate()` via
`ctx->DeferWindowUpdate()` before `ReleaseDeferredWindowUpdate()` in
`H2UnsentResponse::AppendAndDestroySelf()`, and migrated
`finished_streams[i]->ReleaseDeferredWindowUpdate()` via `DeferWindowUpdate()`
after unlocking `_stream_mutex` in `H2Context::FlushPendingData()`.
- Relaxed the server-side `FindOpenStream() == NULL` log in
`H2Context::OnData()` from `LOG(ERROR)` to `RPC_VLOG`.
- Added unit test
`H2UnsentMessage.early_response_migrates_deferred_window_update` covering both
immediate and deferred early-response completion paths.
##########
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:
Fixed in `6605a1a1`: replaced `sctx->Destroy()` with
`ctx->RemoveStreamAndDeferWU(1)` + `removed_sctx->RemoveOneRefForStage2()`.
--
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]
