hjwsm1989 commented on code in PR #3690:
URL: https://github.com/apache/brpc/pull/3690#discussion_r4440820510
##########
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:
Good catch! Fixed in `9284f3f0` (and squashed the branch into a single
commit):
- Added `AddIdleFailRef()` / `ReleaseIdleFailRef()` /
`shall_fail_me_at_idle_timeout()` on `Socket` (`src/brpc/socket.h`,
`src/brpc/socket.cpp`), paired with
`H2StreamContext::HoldDeferredResponseSocket()` and
`H2StreamContext::OnSocketFailed()`.
- During `Server::Stop()`, `Acceptor::StopAccept()` still checks
`shall_fail_me_at_server_stop()` (`false` for H2), so it calls
`ReleaseAdditionalReference()` and preserves graceful draining while the peer
is active.
- When `idle_timeout_sec` expires (`Socket::ReleaseReferenceIfIdle()`),
`shall_fail_me_at_idle_timeout()` is `true` whenever deferred responses are
held on the socket, so `ReleaseReferenceIfIdle()` calls `SetFailed(EUNUSED,
...)` and triggers `H2StreamContext::OnSocketFailed()`.
- In `H2StreamContext::OnSocketFailed()`, when `error_code != 0`, the stream
is removed via `sctx->_conn_ctx->RemoveStreamAndDeferWU(sctx->stream_id())` and
`removed->RemoveOneRefForStage2()` before releasing `sock`, immediately
cleaning up the buffered `_pending_data` / `_pending_trailer` and dropping the
socket reference so `Server::Join()` unblocks.
- Added a separate regression test
`HttpTest.http2_idle_timeout_cleans_up_pending_response` in
`test/brpc_http_rpc_protocol_unittest.cpp` covering both idle expiration while
the server is running and idle expiration after `Server::Stop()` /
`Server::Join()` when a peer advertises a zero stream window and withholds
`WINDOW_UPDATE` / `RST_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]
