chenBright commented on code in PR #3690:
URL: https://github.com/apache/brpc/pull/3690#discussion_r4440456898
##########
src/brpc/policy/http2_rpc_protocol.cpp:
##########
@@ -2012,57 +2162,129 @@ 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);
+ } else {
+ bthread_id_t deferred_notify_id = INVALID_BTHREAD_ID;
+ if (bthread_id_create(
+ &deferred_notify_id, sctx,
+ H2StreamContext::OnSocketFailed) == 0) {
+ // Keep the Socket alive until the deferred response
finishes
+ // draining (or the stream/connection terminates) so
Server::Stop
+ // does not reclaim the Socket while waiting for
WINDOW_UPDATE.
+ sctx->HoldDeferredResponseSocket(socket,
deferred_notify_id);
+ socket->NotifyOnFailed(deferred_notify_id);
Review Comment:
With `idle_timeout_sec = 1`, let a peer advertise a zero stream window,
issue a request producing a nonempty response, and keep TCP open without
sending `WINDOW_UPDATE` or `RST_STREAM`. This new reference keeps the Socket
alive indefinitely: `ReleaseReferenceIfIdle()` only calls
`ReleaseAdditionalReference()` for H2 sockets and never triggers
`OnSocketFailed`. The buffered response survives idle expiration, and
`Server::Join()` can wait indefinitely on this connection. Please add
idle-expiration cleanup for deferred responses while preserving graceful
draining, and cover this case separately from the successful-drain test.
--
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]
