hjwsm1989 commented on code in PR #3690:
URL: https://github.com/apache/brpc/pull/3690#discussion_r4438821558
##########
src/brpc/policy/http2_rpc_protocol.cpp:
##########
@@ -675,6 +728,9 @@ H2ParseResult H2Context::OnHeaders(
tmp_sctx.Init(this, frame_head.stream_id);
tmp_sctx.OnHeaders(it, frame_head, frag_size, pad_length);
return MakeH2Message(nullptr);
+ } else if (remote_ended) {
+ LOG(ERROR) << "Stream already ended remotely, stream_id=" <<
frame_head.stream_id;
+ return MakeH2Error(H2_STREAM_CLOSED_ERROR,
frame_head.stream_id);
Review Comment:
Fixed in 4a7dd66c: Added `H2Context::FindHeaderStream` which marks
`sctx->_rejected_error = H2_STREAM_CLOSED_ERROR` under `_stream_mutex` (and
clears any pending response data/trailer) when `is_server_side() &&
sctx->_remote_ended`, while returning `sctx` so `sctx->OnHeaders` /
`sctx->OnContinuation` consumes the frame payload from `it` and decodes HPACK
without mutating `sctx->header()`, `_parsed_length`, or `_state`, then returns
`MakeH2Error(H2_STREAM_CLOSED_ERROR, stream_id())` once `H2_FLAGS_END_HEADERS`
is reached. Also added unit test
`H2UnsentMessage.rejected_headers_on_remote_ended_stream_preserves_frame_and_hpack`.
##########
src/brpc/policy/http2_rpc_protocol.cpp:
##########
@@ -953,15 +1016,31 @@ H2ParseResult H2StreamContext::OnEndStream() {
return MakeH2Error(H2_PROTOCOL_ERROR);
}
#endif
- H2StreamContext* sctx = _conn_ctx->RemoveStreamAndDeferWU(stream_id());
- if (sctx == nullptr) {
- RPC_VLOG << "Fail to find stream_id=" << stream_id();
- return MakeH2Message(nullptr);
- }
- CHECK_EQ(sctx, this);
+ if (_conn_ctx->is_client_side()) {
+ H2StreamContext* sctx = _conn_ctx->RemoveStreamAndDeferWU(stream_id());
+ if (sctx == nullptr) {
+ RPC_VLOG << "Fail to find stream_id=" << stream_id();
+ return MakeH2Message(nullptr);
+ }
+ CHECK_EQ(sctx, this);
- OnMessageComplete();
- return MakeH2Message(sctx);
+ OnMessageComplete();
+ return MakeH2Message(sctx);
+ } else {
+ if (_remote_ended) {
+ LOG(ERROR) << "Stream already ended remotely in stream_id=" <<
stream_id();
+ return MakeH2Error(H2_PROTOCOL_ERROR);
+ }
+ _remote_ended = true;
+ _conn_ctx->DeferWindowUpdate(ReleaseDeferredWindowUpdate());
+ // Retain sctx in _pending_streams until the response is sent so that
+ // stream-level flow control and WINDOW_UPDATE work for server
responses.
+ // AddOneRefForStage2() adds the reference consumed by
ProcessHttpRequest
+ // (used purely for refcounting; _is_stage2 is only read by HTTP/1.x).
+ AddOneRefForStage2();
Review Comment:
Fixed in 4a7dd66c: Added `req_body.clear()` after protobuf request parsing
on HTTP/2 in `ProcessHttpRequest`, and overrode `H2StreamContext::DestroyImpl`
to clear `body()` when `!read_body_progressively()` before
`HttpContext::DestroyImpl()`.
##########
src/brpc/policy/http2_rpc_protocol.cpp:
##########
@@ -1802,8 +1995,24 @@ void
H2UnsentRequest::DestroyStreamUserData(SocketUniquePtr& sending_sock,
_cntl = nullptr;
if (_stream_id != 0) {
H2Context* ctx =
static_cast<H2Context*>(sending_sock->parsing_context());
- ctx->ClearPendingData(_stream_id);
- ctx->AddAbandonedStream(_stream_id);
+ if (ctx->ClearPendingData(_stream_id)) {
Review Comment:
Fixed in 4a7dd66c: Updated `H2UnsentRequest::DestroyStreamUserData` to call
`ctx->RemoveStreamAndDeferWU(_stream_id)` whenever `error_code != 0 &&
_stream_id != 0`, releasing `sctx` and sending `RST_STREAM(H2_CANCEL)` if the
stream is still active locally (both mid-upload and after upload completes).
Extended
`HttpTest.http2_client_upload_timeout_sends_rst_and_cleans_server_stream` to
cover both cases.
##########
src/brpc/policy/http2_rpc_protocol.cpp:
##########
@@ -1606,80 +1654,229 @@ butil::Status H2Context::TryToInsertClientStream(
ELIMIT, "Pending Stream count exceeds max concurrent stream");
}
- sctx->_remote_window_left.store(_remote_settings.stream_window_size,
- butil::memory_order_relaxed);
const int64_t conn_window =
_remote_window_left.load(butil::memory_order_relaxed);
- const int64_t stream_window =
- sctx->_remote_window_left.load(butil::memory_order_relaxed);
+ const int64_t stream_window = _remote_settings.stream_window_size;
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;
- if (FLAGS_socket_max_unwritten_bytes > 0) {
- const auto limit =
- static_cast<size_t>(FLAGS_socket_max_unwritten_bytes);
+ const int64_t max_pending = GetH2MaxPendingDataBytes();
+ if (max_pending > 0) {
+ const auto limit = static_cast<size_t>(max_pending);
+ const size_t cur_pending =
+ _pending_data_size.load(butil::memory_order_relaxed);
// Check and reserve pending bytes under the same lock. Otherwise,
// concurrent requests may all observe available capacity before any
// of them adds its unsent DATA.
- if (_pending_data_size > limit ||
- pending_size > limit - _pending_data_size) {
+ if (cur_pending > limit || pending_size > limit - cur_pending) {
return butil::Status(EOVERCROWDED,
"Too much pending HTTP/2 request data");
}
}
// Mutate stream and window state only after all failure checks above.
+ sctx->_conn_ctx = this;
+ sctx->_stream_id = stream_id;
+ sctx->_remote_window_left.store(stream_window,
+ butil::memory_order_relaxed);
_pending_streams[stream_id] = sctx;
+ _volatile_pending_stream_size.store(
+ _pending_streams.size(), butil::memory_order_relaxed);
if (!data.empty()) {
CHECK(sctx->_pending_data.empty());
sctx->_pending_data = data;
- _pending_data_size += data.size();
+ AddPendingDataSize(data.size());
AppendPendingDataLocked(sctx, out);
}
return butil::Status::OK();
}
-void H2Context::ClearPendingData(int stream_id) {
+int H2Context::CheckServerStreamBeforeHeaders(
+ int stream_id, size_t data_size, H2Settings* out_remote_settings) {
+ H2StreamContext* overcrowded_sctx = nullptr;
+ {
+ std::unique_lock<butil::Mutex> mu(_stream_mutex);
+ H2StreamContext** psctx = _pending_streams.seek(stream_id);
+ if (psctx == nullptr) {
+ return -1;
+ }
+ H2StreamContext* sctx = *psctx;
+ const int64_t conn_window =
+ _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 =
+ _pending_data_size.load(butil::memory_order_relaxed);
+ if (cur_pending > limit || pending_size > limit - cur_pending) {
+ _pending_streams.erase(stream_id);
+ _volatile_pending_stream_size.store(
+ _pending_streams.size(), butil::memory_order_relaxed);
+ SubPendingDataSize(sctx->_pending_data.size());
+ overcrowded_sctx = sctx;
+ }
+ }
+ if (overcrowded_sctx == nullptr && out_remote_settings != nullptr) {
+ *out_remote_settings = _remote_settings;
+ }
+ }
+ if (overcrowded_sctx != nullptr) {
+ DeferWindowUpdate(overcrowded_sctx->ReleaseDeferredWindowUpdate());
+ overcrowded_sctx->RemoveOneRefForStage2();
+ return 1;
+ }
+ return 0;
+}
+
+void H2Context::CommitServerResponse(int stream_id,
+ const butil::IOBuf& data,
+ butil::IOBuf* trailer_headers,
+ butil::IOBuf* out) {
+ H2StreamContext* finished_sctx = nullptr;
+ {
+ std::unique_lock<butil::Mutex> mu(_stream_mutex);
+ H2StreamContext** psctx = _pending_streams.seek(stream_id);
+ if (psctx == nullptr) {
+ return;
+ }
+ H2StreamContext* sctx = *psctx;
+ sctx->_response_committed = true;
Review Comment:
Fixed in 4a7dd66c: Unified server response admission check, HPACK header
encoding, and response commit under a single `ctx->_stream_mutex` critical
section in `H2UnsentResponse::AppendAndDestroySelf`, removing
`CheckServerStreamBeforeHeaders` and `CommitServerResponse`.
--
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]
