wasphin commented on code in PR #3690:
URL: https://github.com/apache/brpc/pull/3690#discussion_r4436227165
##########
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:
[P2] Retaining the stream now also retains its receive-side request body.
For protobuf requests, `ProcessHttpRequest()` parses from `req_body` without
consuming its IOBuf. Previously, resetting `imsg_guard` before the service call
released that storage; with this additional map-owned lifetime, the original
body remains throughout business processing and any deferred response, outside
pending-DATA accounting. Please release receive-side storage once request
processing no longer needs it, while retaining the necessary stream/send state.
This is unnecessary retention rather than a claim of a permanent leak.
##########
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:
[P2] Please consume rejected header blocks before continuing the connection.
This new `remote_ended` branch returns `STREAM_CLOSED` before consuming the
HEADERS payload. `Consume()` sends RST_STREAM and returns success with no
message, so `ParseH2Message()` continues at the HPACK payload and interprets it
as another frame header. Header compression state is also left unprocessed.
Please preserve frame alignment and HPACK state when rejecting the stream,
including fragmented blocks; the corresponding `OnContinuation()` branch needs
the same consideration.
##########
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:
[P2] Please also notify the peer when cancellation happens after the request
upload completes. Sending CANCEL only when `ClearPendingData()` finds outgoing
request bytes misses an active stream whose response is blocked by flow
control. For example, with the accepted zero client receive stream window, the
server can send response HEADERS but retain all DATA. If the client then times
out, this branch is skipped and only `AddAbandonedStream()` runs, leaving the
server's retained stream/response without cancellation notification. Please
base local timeout/cancellation handling on whether the stream is still active,
rather than whether request bytes remain, and avoid sending a duplicate reset
for streams already completed or reset by the peer.
##########
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:
[P2] The admission check and actual buffering can observe different stream
windows. `CheckServerStreamBeforeHeaders()` releases `_stream_mutex` without
reserving capacity. Between that check and this commit, the reader can apply
SETTINGS that reduces the stream window. A response fitting the old window
passes with zero pending bytes, but can then be entirely buffered here after
the window becomes zero, even if it exceeds the pending limit. Commit does not
recheck the bound. Socket write serialization does not exclude the reader
applying SETTINGS. Please keep the admission/reservation/commit contract
consistent with window changes, while preserving HPACK ordering if admission
fails.
--
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]
