chenBright commented on code in PR #3690:
URL: https://github.com/apache/brpc/pull/3690#discussion_r4439530537
##########
src/brpc/policy/http2_rpc_protocol.cpp:
##########
@@ -2012,57 +2152,117 @@ 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;
+ {
+ 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:
Retaining sctx here does not keep the Socket alive. If the response exceeds
the stream window, after the first batch of DATA is written and
HttpResponseSender destroys the Controller, the remaining data exists only in
H2Context. At that point, Server::Stop releases the connection's persistent
reference through Acceptor::StopAccept, reclaiming the Socket and the pending
data, so the client does not receive the complete response or END_STREAM.
Please account for pending responses in connection lifetime management and
shutdown draining, and add a regression test for calling Stop/Join while a
response with a small stream window is waiting for WINDOW_UPDATE.
--
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]
