This is an automated email from the ASF dual-hosted git repository.
wasphin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/brpc.git
The following commit(s) were added to refs/heads/master by this push:
new 69be4dba5 Support flow-controlled HTTP/2 and gRPC server responses
(#3690)
69be4dba5 is described below
commit 69be4dba5ad521908933ebf3b7cd43b791c915fc
Author: huangjun <[email protected]>
AuthorDate: Sun Nov 8 23:37:09 2026 +0800
Support flow-controlled HTTP/2 and gRPC server responses (#3690)
* Support flow-controlled HTTP/2 and gRPC server responses
- Retain server-side H2StreamContext in _pending_streams across OnEndStream
until the response DATA and trailers are fully flushed or the stream is
reset/abandoned, using intrusive Stage 2 reference counting.
- Hold a Socket reference for deferred responses so Server::Stop() drains
pending DATA/trailers gracefully upon WINDOW_UPDATE, while failing idle
connections on idle_timeout_sec expiration to clean up buffered responses.
- Split server response DATA frames according to min(connection_window,
stream_window, max_frame_size) and resume sending upon WINDOW_UPDATE with
round-robin stream fairness (_last_flushed_stream_id).
- Encode gRPC trailer headers via HPacker::EncodeWithoutDynamicTable so
deferring trailers behind pending DATA never desynchronizes the
connection-level HPACK dynamic table across concurrent streams.
- Guard server pending response buffer with -h2_max_pending_data_bytes /
-socket_max_unwritten_bytes (emitting RST_STREAM(ENHANCE_YOUR_CALM) when
overcrowded) and release req_body immediately after request deserialization in
ProcessHttpRequest.
- Send RST_STREAM(H2_CANCEL) and clean up client stream context when an
HTTP/2 RPC is canceled or times out.
- Include RST_STREAM error code and stream_id in client
Controller::ErrorText while preserving Controller::ErrorCode() == EHTTP.
- Expose h2_stream_context_count, h2_stream_reset_count, and
h2_pending_data_bytes bvars in release builds and H2Context::Describe.
* fix: migrate deferred request window update on early H2 response
completion
---------
Co-authored-by: huangjun <[email protected]>
---
src/brpc/details/hpack.cpp | 23 +
src/brpc/details/hpack.h | 9 +
src/brpc/policy/http2_rpc_protocol.cpp | 767 +++++++++++++++++++----------
src/brpc/policy/http2_rpc_protocol.h | 84 +++-
src/brpc/policy/http_rpc_protocol.cpp | 56 ++-
src/brpc/socket.cpp | 11 +-
src/brpc/socket.h | 16 +
test/brpc_grpc_protocol_unittest.cpp | 125 ++++-
test/brpc_h2_unsent_message_unittest.cpp | 821 ++++++++++++++++++++++++++++++-
test/brpc_http_rpc_protocol_unittest.cpp | 419 +++++++++++++++-
10 files changed, 2029 insertions(+), 302 deletions(-)
diff --git a/src/brpc/details/hpack.cpp b/src/brpc/details/hpack.cpp
index 72eced7fe..a4e74b852 100644
--- a/src/brpc/details/hpack.cpp
+++ b/src/brpc/details/hpack.cpp
@@ -748,6 +748,29 @@ void HPacker::Encode(butil::IOBufAppender* out, const
Header& header,
EncodeString<false>(out, header.value, options.encode_value);
}
+void HPacker::EncodeWithoutDynamicTable(butil::IOBufAppender* out,
+ const Header& header,
+ const HPackOptions& options) {
+ CreateStaticTableOnceOrDie();
+ if (options.index_policy != HPACK_NEVER_INDEX_HEADER) {
+ const HeaderAndHashCode hhc = { HeaderHasher()(header), &header };
+ const int index = s_static_table->GetIndexOfHeader(hhc);
+ if (index > 0) {
+ return EncodeInteger(out, 0x80, 7, index);
+ }
+ }
+ const int name_index = s_static_table->GetIndexOfName(header.name);
+ if (options.index_policy == HPACK_NEVER_INDEX_HEADER) {
+ EncodeInteger(out, 0x10, 4, name_index);
+ } else {
+ EncodeInteger(out, 0x00, 4, name_index);
+ }
+ if (name_index == 0) {
+ EncodeString<true>(out, header.name, options.encode_name);
+ }
+ EncodeString<false>(out, header.value, options.encode_value);
+}
+
inline const HPacker::Header* HPacker::HeaderAt(int index) const {
return (index >= _decode_table->start_index())
? _decode_table->HeaderAt(index) : s_static_table->HeaderAt(index);
diff --git a/src/brpc/details/hpack.h b/src/brpc/details/hpack.h
index bd1da3211..0f3e379c3 100644
--- a/src/brpc/details/hpack.h
+++ b/src/brpc/details/hpack.h
@@ -108,6 +108,15 @@ public:
void Encode(butil::IOBufAppender* out, const Header& header)
{ return Encode(out, header, HPackOptions()); }
+ // Encode header without consulting or mutating the dynamic table (uses
+ // static-table indexed representation when matched, or literal
representation
+ // per RFC 7541 section 6.1 / 6.2.2 / 6.2.3). Safe to use for trailers
whose
+ // transmission may be deferred behind flow-controlled DATA frames while
+ // other streams continue to encode HEADERS on the same connection.
+ static void EncodeWithoutDynamicTable(butil::IOBufAppender* out,
+ const Header& header,
+ const HPackOptions& options);
+
// Try to decode at most one Header from source and erase corresponding
// buffer.
// Returns:
diff --git a/src/brpc/policy/http2_rpc_protocol.cpp
b/src/brpc/policy/http2_rpc_protocol.cpp
index eb788d5a7..438603bdc 100644
--- a/src/brpc/policy/http2_rpc_protocol.cpp
+++ b/src/brpc/policy/http2_rpc_protocol.cpp
@@ -52,6 +52,15 @@ DEFINE_bool(h2_hpack_encode_name, false,
"Encode name in HTTP2 headers with huffman encoding");
DEFINE_bool(h2_hpack_encode_value, false,
"Encode value in HTTP2 headers with huffman encoding");
+DEFINE_int64(h2_max_pending_data_bytes, 0,
+ "Max bytes of unsent HTTP/2 DATA buffered per connection due to "
+ "remote flow control. 0 means using -socket_max_unwritten_bytes");
+
+inline int64_t GetH2MaxPendingDataBytes() {
+ return FLAGS_h2_max_pending_data_bytes > 0
+ ? FLAGS_h2_max_pending_data_bytes
+ : FLAGS_socket_max_unwritten_bytes;
+}
DEFINE_bool(h2_ack_ignore_eovercrowded, false,
"Let h2 control-frame replies (PING/SETTINGS acks, RST_STREAM, "
@@ -305,19 +314,6 @@ inline bool AddWindowSize(butil::atomic<int64_t>*
window_size, int64_t diff) {
return true;
}
-inline bool MinusWindowSize(butil::atomic<int64_t>* window_size, int64_t size)
{
- if (window_size->load(butil::memory_order_relaxed) < size) {
- // false negative is OK.
- return false;
- }
- int64_t before_sub = window_size->fetch_sub(size,
butil::memory_order_relaxed);
- if (before_sub < size) {
- window_size->fetch_add(size, butil::memory_order_relaxed);
- return false;
- }
- return true;
-}
-
static H2Context::FrameHandler s_frame_handlers[H2_FRAME_TYPE_MAX + 1];
static pthread_once_t s_frame_handlers_init_once = PTHREAD_ONCE_INIT;
void InitFrameHandlers() {
@@ -348,7 +344,10 @@ H2Context::H2Context(Socket* socket, const Server* server)
, _last_sent_stream_id(1)
, _goaway_stream_id(-1)
, _remote_settings_received(false)
+ , _client_preface_pending(false)
+ , _volatile_pending_stream_size(0)
, _pending_data_size(0)
+ , _last_flushed_stream_id(0)
, _deferred_window_update(0) {
// Stop printing the field which is useless for remote settings.
_remote_settings.connection_window_size = 0;
@@ -379,11 +378,29 @@ H2Context::H2Context(Socket* socket, const Server* server)
}
H2Context::~H2Context() {
+ SubPendingDataSize(_pending_data_size.load(butil::memory_order_relaxed));
for (StreamMap::iterator it = _pending_streams.begin();
it != _pending_streams.end(); ++it) {
- delete it->second;
+ it->second->RemoveOneRefForStage2();
}
_pending_streams.clear();
+ _volatile_pending_stream_size.store(0, butil::memory_order_relaxed);
+}
+
+void H2Context::AddPendingDataSize(size_t n) {
+ if (n > 0) {
+ _pending_data_size.fetch_add(n, butil::memory_order_relaxed);
+ get_h2_bvars()->h2_pending_data_bytes << static_cast<int64_t>(n);
+ }
+}
+
+void H2Context::SubPendingDataSize(size_t n) {
+ if (n > 0) {
+ const size_t prev =
+ _pending_data_size.fetch_sub(n, butil::memory_order_relaxed);
+ CHECK_GE(prev, n);
+ get_h2_bvars()->h2_pending_data_bytes << -static_cast<int64_t>(n);
+ }
}
int H2Context::Init() {
@@ -402,19 +419,28 @@ H2Settings H2Context::remote_settings() const {
}
size_t H2Context::VolatilePendingStreamSize() const {
- std::unique_lock<butil::Mutex> mu(_stream_mutex);
- return _pending_streams.size();
+ return _volatile_pending_stream_size.load(butil::memory_order_relaxed);
+}
+
+H2StreamContext* H2Context::RemoveStreamLocked(int stream_id) {
+ H2StreamContext* sctx = nullptr;
+ if (!_pending_streams.erase(stream_id, &sctx)) {
+ return nullptr;
+ }
+ _volatile_pending_stream_size.store(
+ _pending_streams.size(), butil::memory_order_relaxed);
+ SubPendingDataSize(sctx->_pending_data.size());
+ return sctx;
}
H2StreamContext* H2Context::RemoveStreamAndDeferWU(int stream_id) {
H2StreamContext* sctx = nullptr;
{
std::unique_lock<butil::Mutex> mu(_stream_mutex);
- if (!_pending_streams.erase(stream_id, &sctx)) {
+ sctx = RemoveStreamLocked(stream_id);
+ if (sctx == nullptr) {
return nullptr;
}
- CHECK_GE(_pending_data_size, sctx->_pending_data.size());
- _pending_data_size -= sctx->_pending_data.size();
}
// The remote stream will not send any more data, sending back the
// stream-level WINDOW_UPDATE is pointless, just move the value into
@@ -432,7 +458,9 @@ void H2Context::RemoveGoAwayStreams(
std::unique_lock<butil::Mutex> mu(_stream_mutex);
_goaway_stream_id = goaway_stream_id;
_pending_streams.swap(tmp);
- _pending_data_size = 0;
+ _volatile_pending_stream_size.store(0,
butil::memory_order_relaxed);
+ SubPendingDataSize(
+ _pending_data_size.load(butil::memory_order_relaxed));
}
for (StreamMap::const_iterator it = tmp.begin(); it != tmp.end();
++it) {
out_streams->push_back(it->second);
@@ -447,10 +475,7 @@ void H2Context::RemoveGoAwayStreams(
}
}
for (size_t i = 0; i < out_streams->size(); ++i) {
- CHECK_GE(_pending_data_size,
- (*out_streams)[i]->_pending_data.size());
- _pending_data_size -= (*out_streams)[i]->_pending_data.size();
- _pending_streams.erase((*out_streams)[i]->stream_id());
+ RemoveStreamLocked((*out_streams)[i]->stream_id());
}
}
}
@@ -464,8 +489,46 @@ H2StreamContext* H2Context::FindStream(int stream_id) {
return nullptr;
}
+butil::intrusive_ptr<H2StreamContext> H2Context::FindOpenStream(int stream_id)
{
+ std::unique_lock<butil::Mutex> mu(_stream_mutex);
+ H2StreamContext** psctx = _pending_streams.seek(stream_id);
+ if (psctx == nullptr) {
+ return nullptr;
+ }
+ H2StreamContext* sctx = *psctx;
+ if (is_server_side() &&
+ (sctx->_remote_ended || sctx->_rejected_error != H2_NO_ERROR)) {
+ return nullptr;
+ }
+ // SharedObject's intrusive_ptr_add_ref/intrusive_ptr_release manipulate
+ // the same reference count as AddOneRefForStage2/RemoveOneRefForStage2.
+ return butil::intrusive_ptr<H2StreamContext>(sctx);
+}
+
+butil::intrusive_ptr<H2StreamContext> H2Context::FindHeaderStream(
+ int stream_id) {
+ std::unique_lock<butil::Mutex> mu(_stream_mutex);
+ H2StreamContext** psctx = _pending_streams.seek(stream_id);
+ if (psctx == nullptr) {
+ return nullptr;
+ }
+ H2StreamContext* sctx = *psctx;
+ if (is_server_side() && sctx->_remote_ended &&
+ sctx->_rejected_error == H2_NO_ERROR) {
+ LOG(ERROR) << "Stream already ended remotely, stream_id=" << stream_id;
+ sctx->_rejected_error = H2_STREAM_CLOSED_ERROR;
+ SubPendingDataSize(sctx->_pending_data.size());
+ sctx->_pending_data.clear();
+ sctx->_pending_trailer.clear();
+ }
+ return butil::intrusive_ptr<H2StreamContext>(sctx);
+}
+
int H2Context::TryToInsertStream(int stream_id, H2StreamContext* ctx) {
std::unique_lock<butil::Mutex> mu(_stream_mutex);
+ if (is_server_side() && stream_id > _last_received_stream_id) {
+ _last_received_stream_id = stream_id;
+ }
if (_goaway_stream_id >= 0 && stream_id > _goaway_stream_id) {
return 1;
}
@@ -481,10 +544,15 @@ int H2Context::TryToInsertStream(int stream_id,
H2StreamContext* ctx) {
}
H2StreamContext*& sctx = _pending_streams[stream_id];
if (sctx == nullptr) {
+ if (is_server_side() && stream_id > _last_received_stream_id) {
+ _last_received_stream_id = stream_id;
+ }
// Synchronize creation with SETTINGS_INITIAL_WINDOW_SIZE updates.
ctx->_remote_window_left.store(_remote_settings.stream_window_size,
butil::memory_order_relaxed);
sctx = ctx;
+ _volatile_pending_stream_size.store(
+ _pending_streams.size(), butil::memory_order_relaxed);
return 0;
}
return -1;
@@ -565,6 +633,7 @@ ParseResult H2Context::Consume(
SerializeFrameHead(rstbuf, 4, H2_FRAME_RST_STREAM,
0, h2_res.stream_id());
SaveUint32(rstbuf + FRAME_HEAD_SIZE, h2_res.error());
+ get_h2_bvars()->h2_stream_reset_count << 1;
if (WriteAck(_socket, rstbuf, sizeof(rstbuf)) != 0) {
LOG(WARNING) << "Fail to send RST_STREAM to " << *_socket;
return MakeParseError(PARSE_ERROR_ABSOLUTELY_WRONG);
@@ -572,7 +641,7 @@ ParseResult H2Context::Consume(
H2StreamContext* sctx = RemoveStreamAndDeferWU(h2_res.stream_id());
if (sctx) {
if (is_server_side()) {
- delete sctx;
+ sctx->RemoveOneRefForStage2();
return MakeMessage(nullptr);
} else {
sctx->header().set_status_code(
@@ -633,6 +702,7 @@ H2ParseResult H2Context::OnHeaders(
return MakeH2Error(H2_FRAME_SIZE_ERROR);
}
frag_size -= pad_length;
+ butil::intrusive_ptr<H2StreamContext> sctx_holder;
H2StreamContext* sctx = nullptr;
if (is_server_side() &&
frame_head.stream_id > _last_received_stream_id) { // new stream
@@ -641,16 +711,15 @@ H2ParseResult H2Context::OnHeaders(
<< " created by client is not odd";
return MakeH2Error(H2_PROTOCOL_ERROR);
}
- _last_received_stream_id = frame_head.stream_id;
sctx = new H2StreamContext(_socket->is_read_progressive());
sctx->Init(this, frame_head.stream_id);
const int rc = TryToInsertStream(frame_head.stream_id, sctx);
if (rc < 0) {
- delete sctx;
+ sctx->RemoveOneRefForStage2();
LOG(ERROR) << "Fail to insert existing stream_id=" <<
frame_head.stream_id;
return MakeH2Error(H2_PROTOCOL_ERROR);
} else if (rc == 2) {
- delete sctx;
+ sctx->RemoveOneRefForStage2();
LOG_EVERY_SECOND(WARNING)
<< "Refused stream_id=" << frame_head.stream_id
<< " since concurrent streams reached max_concurrent_streams="
@@ -662,11 +731,13 @@ H2ParseResult H2Context::OnHeaders(
// and REFUSED_STREAM lets a compliant client retry later.
return MakeH2Error(H2_REFUSED_STREAM, frame_head.stream_id);
} else if (rc > 0) {
- delete sctx;
+ sctx->RemoveOneRefForStage2();
return MakeH2Error(H2_REFUSED_STREAM);
}
+ sctx_holder.reset(sctx);
} else {
- sctx = FindStream(frame_head.stream_id);
+ sctx_holder = FindHeaderStream(frame_head.stream_id);
+ sctx = sctx_holder.get();
if (sctx == nullptr) {
if (is_client_side()) {
RPC_VLOG << "Fail to find stream_id=" << frame_head.stream_id;
@@ -692,10 +763,12 @@ bool H2StreamContext::HeaderFragmentTooLarge() const {
H2ParseResult H2StreamContext::OnHeaders(
butil::IOBufBytesIterator& it, const H2FrameHead& frame_head,
uint32_t frag_size, uint8_t pad_length) {
- _parsed_length += FRAME_HEAD_SIZE + frame_head.payload_size;
+ if (_rejected_error == H2_NO_ERROR) {
+ _parsed_length += FRAME_HEAD_SIZE + frame_head.payload_size;
#if defined(BRPC_H2_STREAM_STATE)
- SetState(H2_STREAM_OPEN);
+ SetState(H2_STREAM_OPEN);
#endif
+ }
// A new HEADERS block (which may be an initial request header set, or
// trailing headers on the same stream) starts here. The decoded header
// list budget is per header block (RFC 7540 section 10.5.1), so reset the
@@ -740,7 +813,8 @@ H2ParseResult H2StreamContext::OnHeaders(
}
return MakeH2Message(nullptr);
} else {
- if (frame_head.flags & H2_FLAGS_END_STREAM) {
+ if (_rejected_error == H2_NO_ERROR &&
+ (frame_head.flags & H2_FLAGS_END_STREAM)) {
// Delay calling OnEndStream() in OnContinuation()
_stream_ended = true;
}
@@ -750,7 +824,8 @@ H2ParseResult H2StreamContext::OnHeaders(
H2ParseResult H2Context::OnContinuation(
butil::IOBufBytesIterator& it, const H2FrameHead& frame_head) {
- H2StreamContext* sctx = FindStream(frame_head.stream_id);
+ butil::intrusive_ptr<H2StreamContext> sctx =
+ FindHeaderStream(frame_head.stream_id);
if (sctx == nullptr) {
if (is_client_side()) {
RPC_VLOG << "Fail to find stream_id=" << frame_head.stream_id;
@@ -769,7 +844,9 @@ H2ParseResult H2Context::OnContinuation(
H2ParseResult H2StreamContext::OnContinuation(
butil::IOBufBytesIterator& it, const H2FrameHead& frame_head) {
- _parsed_length += FRAME_HEAD_SIZE + frame_head.payload_size;
+ if (_rejected_error == H2_NO_ERROR) {
+ _parsed_length += FRAME_HEAD_SIZE + frame_head.payload_size;
+ }
it.append_and_forward(&_remaining_header_fragment,
frame_head.payload_size);
// A header block may span many CONTINUATION frames; ConsumeHeaders()
// drains complete fields, so the fragment only buffers one incomplete
@@ -825,7 +902,8 @@ H2ParseResult H2Context::OnData(
return MakeH2Error(H2_FRAME_SIZE_ERROR);
}
frag_size -= pad_length;
- H2StreamContext* sctx = FindStream(frame_head.stream_id);
+ butil::intrusive_ptr<H2StreamContext> sctx =
+ FindOpenStream(frame_head.stream_id);
if (sctx == nullptr) {
// If a DATA frame is received whose stream is not in "open" or
"half-closed (local)" state,
// the recipient MUST respond with a stream error (Section 5.4.2) of
type STREAM_CLOSED.
@@ -835,8 +913,13 @@ H2ParseResult H2Context::OnData(
tmp_sctx.OnData(it, frame_head, frag_size, pad_length);
DeferWindowUpdate(tmp_sctx.ReleaseDeferredWindowUpdate());
- LOG(ERROR) << "Fail to find stream_id=" << frame_head.stream_id;
- return MakeH2Error(H2_STREAM_CLOSED_ERROR, frame_head.stream_id);
+ if (is_server_side()) {
+ RPC_VLOG << "Fail to find stream_id=" << frame_head.stream_id;
+ return MakeH2Error(H2_STREAM_CLOSED_ERROR, frame_head.stream_id);
+ } else {
+ RPC_VLOG << "Fail to find stream_id=" << frame_head.stream_id;
+ return MakeH2Message(nullptr);
+ }
}
return sctx->OnData(it, frame_head, frag_size, pad_length);
}
@@ -904,7 +987,7 @@ H2ParseResult H2Context::OnResetStream(
return MakeH2Error(H2_FRAME_SIZE_ERROR);
}
const H2Error h2_error = static_cast<H2Error>(LoadUint32(it));
- H2StreamContext* sctx = FindStream(frame_head.stream_id);
+ H2StreamContext* sctx = RemoveStreamAndDeferWU(frame_head.stream_id);
if (sctx == nullptr) {
RPC_VLOG << "Fail to find stream_id=" << frame_head.stream_id;
return MakeH2Message(nullptr);
@@ -918,25 +1001,25 @@ H2ParseResult H2StreamContext::OnResetStream(
#if defined(BRPC_H2_STREAM_STATE)
if (state() == H2_STREAM_OPEN) {
SetState(H2_STREAM_HALF_CLOSED_REMOTE);
- } else if (state() == H2_STREAM_HALF_CLOSED_LOCAL) {
+ } else if (state() == H2_STREAM_HALF_CLOSED_LOCAL ||
+ state() == H2_STREAM_HALF_CLOSED_REMOTE) {
SetState(H2_STREAM_CLOSED);
} else {
LOG(ERROR) << "Invalid state=" << H2StreamState2Str(_state)
<< " in stream_id=" << stream_id();
+ RemoveOneRefForStage2();
return MakeH2Error(H2_PROTOCOL_ERROR);
}
#endif
- H2StreamContext* sctx = _conn_ctx->RemoveStreamAndDeferWU(stream_id());
- if (sctx == nullptr) {
- LOG(ERROR) << "Fail to find stream_id=" << stream_id();
- return MakeH2Error(H2_PROTOCOL_ERROR);
- }
+ get_h2_bvars()->h2_stream_reset_count << 1;
if (_conn_ctx->is_client_side()) {
- sctx->header().set_status_code(H2ErrorToStatusCode(h2_error));
- return MakeH2Message(sctx);
+ _rst_received = true;
+ _rst_error = h2_error;
+ header().set_status_code(H2ErrorToStatusCode(h2_error));
+ return MakeH2Message(this);
} else {
- // No need to process the request.
- delete sctx;
+ // No need to process the request. Release _pending_streams's ref.
+ RemoveOneRefForStage2();
return MakeH2Message(nullptr);
}
}
@@ -953,15 +1036,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();
+ OnMessageComplete();
+ return MakeH2Message(this);
+ }
}
H2ParseResult H2Context::OnSettings(
@@ -1148,7 +1247,7 @@ H2ParseResult H2Context::OnWindowUpdate(
LOG(ERROR) << "Invalid stream-level window_size_increment=" <<
inc
<< " to remote_window_left="
<<
(*psctx)->_remote_window_left.load(butil::memory_order_relaxed);
- return MakeH2Error(H2_FLOW_CONTROL_ERROR);
+ return MakeH2Error(H2_FLOW_CONTROL_ERROR,
frame_head.stream_id);
}
}
if (!FlushPendingData(frame_head.stream_id)) {
@@ -1161,21 +1260,23 @@ H2ParseResult H2Context::OnWindowUpdate(
void H2Context::Describe(std::ostream& os, const DescribeOptions& opt) const {
H2Settings remote_settings;
bool remote_settings_received = false;
+ int last_received_stream_id = -1;
size_t pending_stream_size = 0;
size_t pending_data_size = 0;
{
std::unique_lock<butil::Mutex> mu(_stream_mutex);
remote_settings = _remote_settings;
remote_settings_received = _remote_settings_received;
+ last_received_stream_id = _last_received_stream_id;
pending_stream_size = _pending_streams.size();
- pending_data_size = _pending_data_size;
+ pending_data_size =
_pending_data_size.load(butil::memory_order_relaxed);
}
if (opt.verbose) {
os << '\n';
}
const char sep = (opt.verbose ? '\n' : ' ');
os << "conn_state=" << H2ConnectionState2Str(_conn_state);
- os << sep << "last_received_stream_id=" << _last_received_stream_id
+ os << sep << "last_received_stream_id=" << last_received_stream_id
<< sep << "last_sent_stream_id=" << _last_sent_stream_id;
os << sep << "deferred_window_update="
<< _deferred_window_update.load(butil::memory_order_relaxed)
@@ -1287,7 +1388,7 @@ inline void H2Context::ClearAbandonedStreams() {
mu.unlock();
H2StreamContext* sctx = RemoveStreamAndDeferWU(stream_id);
if (sctx != nullptr) {
- delete sctx;
+ sctx->RemoveOneRefForStage2();
}
mu.lock();
}
@@ -1301,15 +1402,46 @@ H2StreamContext::H2StreamContext(bool
read_body_progressively)
#endif
, _stream_id(0)
, _stream_ended(false)
+ , _remote_ended(false)
+ , _response_committed(false)
+ , _rst_received(false)
+ , _rst_error(H2_NO_ERROR)
, _remote_window_left(0)
, _deferred_window_update(0)
, _correlation_id(INVALID_BTHREAD_ID.value)
, _decoded_header_list_size(0)
- , _rejected_error(H2_NO_ERROR) {
+ , _rejected_error(H2_NO_ERROR)
+ , _deferred_response_id(INVALID_BTHREAD_ID.value) {
header().set_version(2, 0);
-#ifndef NDEBUG
get_h2_bvars()->h2_stream_context_count << 1;
-#endif
+}
+
+int H2StreamContext::OnSocketFailed(bthread_id_t id, void* data, int
error_code) {
+ butil::intrusive_ptr<H2StreamContext> sctx(
+ static_cast<H2StreamContext*>(data), false);
+ sctx->_deferred_response_id.store(
+ INVALID_BTHREAD_ID.value, butil::memory_order_release);
+ SocketUniquePtr sock(std::move(sctx->_deferred_response_sock));
+ if (sock != NULL) {
+ sock->ReleaseIdleFailRef();
+ }
+ int rc = 0;
+ if (error_code != 0 && sctx->_conn_ctx != NULL) {
+ H2StreamContext* removed =
+ sctx->_conn_ctx->RemoveStreamAndDeferWU(sctx->stream_id());
+ if (removed != NULL) {
+ if (sock != NULL && !sock->Failed()) {
+ // Socket::NotifyOnFailed() failed to register `id' in the
+ // notification list (e.g. EAGAIN) while the Socket itself is
+ // still healthy. Return `error_code' so AppendAndDestroySelf()
+ // can explicitly reset the stream via RST_STREAM.
+ rc = error_code;
+ }
+ removed->RemoveOneRefForStage2();
+ }
+ }
+ bthread_id_unlock_and_destroy(id);
+ return rc;
}
void H2StreamContext::Init(H2Context* conn_ctx, int stream_id) {
@@ -1320,9 +1452,7 @@ void H2StreamContext::Init(H2Context* conn_ctx, int
stream_id) {
}
H2StreamContext::~H2StreamContext() {
-#ifndef NDEBUG
get_h2_bvars()->h2_stream_context_count << -1;
-#endif
}
#if defined(BRPC_H2_STREAM_STATE)
@@ -1509,51 +1639,6 @@ static void PackH2Headers(butil::IOBuf* out,
}
}
-static void PackH2Message(butil::IOBuf* out,
- butil::IOBuf& headers,
- butil::IOBuf& trailer_headers,
- const butil::IOBuf& data,
- int stream_id,
- H2Context* conn_ctx) {
- const H2Settings& remote_settings = conn_ctx->remote_settings();
- char headbuf[FRAME_HEAD_SIZE];
- PackH2Headers(out, headers, stream_id, remote_settings.max_frame_size,
- data.empty() && trailer_headers.empty());
- if (!data.empty()) {
- H2FrameHead data_head = {0, H2_FRAME_DATA, 0, stream_id};
- butil::IOBufBytesIterator it(data);
- while (it.bytes_left()) {
- if (it.bytes_left() <= remote_settings.max_frame_size) {
- data_head.payload_size = it.bytes_left();
- if (trailer_headers.empty()) {
- data_head.flags |= H2_FLAGS_END_STREAM;
- }
- } else {
- data_head.payload_size = remote_settings.max_frame_size;
- }
- SerializeFrameHead(headbuf, data_head);
- out->append(headbuf, FRAME_HEAD_SIZE);
- it.append_and_forward(out, data_head.payload_size);
- }
- }
- if (!trailer_headers.empty()) {
- H2FrameHead headers_head = {
- (uint32_t)trailer_headers.size(), H2_FRAME_HEADERS, 0, stream_id};
- headers_head.flags |= H2_FLAGS_END_STREAM;
- headers_head.flags |= H2_FLAGS_END_HEADERS;
- SerializeFrameHead(headbuf, headers_head);
- out->append(headbuf, sizeof(headbuf));
- out->append(butil::IOBuf::Movable(trailer_headers));
- }
- const int64_t conn_wu = conn_ctx->ReleaseDeferredWindowUpdate();
- if (conn_wu > 0) {
- char winbuf[FRAME_HEAD_SIZE + 4];
- SerializeFrameHead(winbuf, 4, H2_FRAME_WINDOW_UPDATE, 0, 0);
- SaveUint32(winbuf + FRAME_HEAD_SIZE, conn_wu);
- out->append(winbuf, sizeof(winbuf));
- }
-}
-
void H2Context::AppendPendingDataLocked(H2StreamContext* sctx,
butil::IOBuf* out) {
CHECK(sctx != nullptr);
@@ -1579,14 +1664,19 @@ void
H2Context::AppendPendingDataLocked(H2StreamContext* sctx,
H2FrameHead data_head = {
static_cast<uint32_t>(payload_size), H2_FRAME_DATA, 0,
sctx->stream_id()};
- if (payload_size == sctx->_pending_data.size()) {
+ if (payload_size == sctx->_pending_data.size() &&
+ sctx->_pending_trailer.empty()) {
data_head.flags |= H2_FLAGS_END_STREAM;
}
SerializeFrameHead(headbuf, data_head);
out->append(headbuf, sizeof(headbuf));
sctx->_pending_data.cutn(out, payload_size);
- CHECK_GE(_pending_data_size, payload_size);
- _pending_data_size -= payload_size;
+ SubPendingDataSize(payload_size);
+ }
+ if (sctx->_pending_data.empty() && !sctx->_pending_trailer.empty()) {
+ PackH2Headers(out, sctx->_pending_trailer, sctx->stream_id(),
+ max_frame_size, true /* end_stream */);
+ sctx->_pending_trailer.clear();
}
}
@@ -1594,6 +1684,12 @@ butil::Status H2Context::TryToInsertClientStream(
int stream_id, H2StreamContext* sctx, const butil::IOBuf& data,
butil::IOBuf* out) {
std::unique_lock<butil::Mutex> mu(_stream_mutex);
+ return TryToInsertClientStreamLocked(stream_id, sctx, data, out);
+}
+
+butil::Status H2Context::TryToInsertClientStreamLocked(
+ int stream_id, H2StreamContext* sctx, const butil::IOBuf& data,
+ butil::IOBuf* out) {
if (_goaway_stream_id >= 0 && stream_id > _goaway_stream_id) {
return butil::Status(ELOGOFF, "the connection just issued GOAWAY");
}
@@ -1606,12 +1702,9 @@ 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(
@@ -1619,68 +1712,133 @@ butil::Status H2Context::TryToInsertClientStream(
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) {
+bool H2Context::ClearPendingData(int stream_id) {
std::unique_lock<butil::Mutex> mu(_stream_mutex);
H2StreamContext** psctx = _pending_streams.seek(stream_id);
if (psctx == nullptr) {
- return;
+ return false;
}
- CHECK_GE(_pending_data_size, (*psctx)->_pending_data.size());
- _pending_data_size -= (*psctx)->_pending_data.size();
+ const bool had_pending = !(*psctx)->_pending_data.empty() ||
+ !(*psctx)->_pending_trailer.empty();
+ SubPendingDataSize((*psctx)->_pending_data.size());
(*psctx)->_pending_data.clear();
+ (*psctx)->_pending_trailer.clear();
+ return had_pending;
}
bool H2Context::PendingDataOvercrowded() const {
- std::unique_lock<butil::Mutex> mu(_stream_mutex);
- return FLAGS_socket_max_unwritten_bytes > 0 &&
- _pending_data_size >=
- static_cast<size_t>(FLAGS_socket_max_unwritten_bytes);
+ const int64_t max_pending = GetH2MaxPendingDataBytes();
+ return max_pending > 0 &&
+ _pending_data_size.load(butil::memory_order_relaxed) >=
+ static_cast<size_t>(max_pending);
}
bool H2Context::FlushPendingData(int stream_id) {
butil::IOBuf out;
+ std::vector<H2StreamContext*> finished_streams;
{
std::unique_lock<butil::Mutex> mu(_stream_mutex);
if (stream_id != 0) {
H2StreamContext** psctx = _pending_streams.seek(stream_id);
if (psctx != nullptr) {
- AppendPendingDataLocked(*psctx, &out);
+ H2StreamContext* sctx = *psctx;
+ if (sctx->_rejected_error == H2_NO_ERROR &&
+ (!sctx->_pending_data.empty() ||
+ !sctx->_pending_trailer.empty())) {
+ AppendPendingDataLocked(sctx, &out);
+ if (is_server_side() && sctx->_response_committed &&
+ sctx->_pending_data.empty() &&
+ sctx->_pending_trailer.empty()) {
+ RemoveStreamLocked(stream_id);
+ finished_streams.push_back(sctx);
+ }
+ }
}
} else {
+ // Round-Robin at connection-level WINDOW_UPDATE granularity:
+ // rotate the starting stream after _last_flushed_stream_id so a
+ // lower-ID stream cannot permanently starve higher-ID streams
+ // across consecutive WINDOW_UPDATE frames.
+ std::vector<H2StreamContext*> pending_list;
for (StreamMap::const_iterator it = _pending_streams.begin();
it != _pending_streams.end(); ++it) {
+ H2StreamContext* sctx = it->second;
+ if (sctx->_rejected_error == H2_NO_ERROR &&
+ (!sctx->_pending_data.empty() ||
+ !sctx->_pending_trailer.empty())) {
+ pending_list.push_back(sctx);
+ }
+ }
+ if (pending_list.size() > 1) {
+ std::sort(pending_list.begin(), pending_list.end(),
+ [](const H2StreamContext* a, const H2StreamContext*
b) {
+ return a->stream_id() < b->stream_id();
+ });
+ }
+ size_t start_idx = 0;
+ for (size_t i = 0; i < pending_list.size(); ++i) {
+ if (pending_list[i]->stream_id() > _last_flushed_stream_id) {
+ start_idx = i;
+ break;
+ }
+ }
+ for (size_t i = 0; i < pending_list.size(); ++i) {
if (_remote_window_left.load(butil::memory_order_relaxed) <=
0) {
break;
}
- AppendPendingDataLocked(it->second, &out);
+ H2StreamContext* sctx =
+ pending_list[(start_idx + i) % pending_list.size()];
+ const size_t before_size = sctx->_pending_data.size();
+ AppendPendingDataLocked(sctx, &out);
+ if (sctx->_pending_data.size() < before_size) {
+ _last_flushed_stream_id = sctx->stream_id();
+ }
+ if (is_server_side() && sctx->_response_committed &&
+ sctx->_pending_data.empty() &&
+ sctx->_pending_trailer.empty()) {
+ RemoveStreamLocked(sctx->stream_id());
+ finished_streams.push_back(sctx);
+ }
}
}
}
- return out.empty() || WriteAck(_socket, &out) == 0;
+ const bool rc = out.empty() || WriteAck(_socket, &out) == 0;
+ for (size_t i = 0; i < finished_streams.size(); ++i) {
+ DeferWindowUpdate(finished_streams[i]->ReleaseDeferredWindowUpdate());
+ finished_streams[i]->RemoveOneRefForStage2();
+ }
+ return rc;
}
H2UnsentRequest::H2UnsentRequest(Controller* c)
@@ -1688,15 +1846,11 @@ H2UnsentRequest::H2UnsentRequest(Controller* c)
, _size(0)
, _stream_id(0)
, _cntl(c) {
-#ifndef NDEBUG
get_h2_bvars()->h2_unsent_request_count << 1;
-#endif
}
H2UnsentRequest::~H2UnsentRequest() {
-#ifndef NDEBUG
get_h2_bvars()->h2_unsent_request_count << -1;
-#endif
}
H2UnsentRequest* H2UnsentRequest::New(Controller* c) {
@@ -1802,8 +1956,21 @@ 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);
+ H2StreamContext* sctx = ctx->RemoveStreamAndDeferWU(_stream_id);
+ if (sctx != NULL) {
+ // The stream is still active locally (neither completed via
+ // OnEndStream nor reset by the peer). Notify the peer with
+ // RST_STREAM(CANCEL) so it can release any retained stream
+ // or pending response DATA.
+ sctx->RemoveOneRefForStage2();
+ char rstbuf[FRAME_HEAD_SIZE + 4];
+ SerializeFrameHead(rstbuf, 4, H2_FRAME_RST_STREAM, 0,
_stream_id);
+ SaveUint32(rstbuf + FRAME_HEAD_SIZE, H2_CANCEL);
+ get_h2_bvars()->h2_stream_reset_count << 1;
+ if (WriteAck(sending_sock.get(), rstbuf, sizeof(rstbuf)) != 0)
{
+ LOG(WARNING) << "Fail to send RST_STREAM to " <<
*sending_sock;
+ }
+ }
}
}
}
@@ -1823,9 +1990,14 @@ H2UnsentRequest::AppendAndDestroySelf(butil::IOBuf* out,
Socket* socket) {
if (socket == nullptr) {
return butil::Status::OK();
}
- H2Context* ctx = static_cast<H2Context*>(socket->parsing_context());
+ // Although the critical section looks huge, it should rarely be contended
+ // since timeout of RPC is much larger than the delay of sending.
+ std::unique_lock<butil::Mutex> mu(_mutex);
+ if (_cntl == nullptr) {
+ return butil::Status(ECANCELED, "The RPC was already failed");
+ }
- // Create a http2 stream and store correlation_id in.
+ H2Context* ctx = static_cast<H2Context*>(socket->parsing_context());
if (ctx == nullptr) {
CHECK(socket->CreatedByConnect());
ctx = new H2Context(socket, nullptr);
@@ -1833,80 +2005,72 @@ H2UnsentRequest::AppendAndDestroySelf(butil::IOBuf*
out, Socket* socket) {
delete ctx;
return butil::Status(EINTERNAL, "Fail to init H2Context");
}
+ ctx->_client_preface_pending = true;
socket->initialize_parsing_context(&ctx);
-
- // Append client connection preface
- out->append(H2_CONNECTION_PREFACE_PREFIX,
- H2_CONNECTION_PREFACE_PREFIX_SIZE);
-
- char settingsbuf[FRAME_HEAD_SIZE + H2_SETTINGS_MAX_BYTE_SIZE +
- FRAME_HEAD_SIZE + 4/*for WU*/];
- const size_t nb = SerializeH2SettingsFrameAndWU(
- ctx->_unack_local_settings, settingsbuf);
- out->append(settingsbuf, nb);
}
- // Although the critical section looks huge, it should rarely be contended
- // since timeout of RPC is much larger than the delay of sending.
- std::unique_lock<butil::Mutex> mu(_mutex);
- if (_cntl == nullptr) {
- return butil::Status(ECANCELED, "The RPC was already failed");
- }
-
- const int id = ctx->AllocateClientStreamId();
- if (id < 0) {
- // The RPC should be failed and retried.
- // Note that the socket should not be SetFailed() which may affect
- // other RPC successfully sent requests and waiting for responses.
- RPC_VLOG << "Fail to allocate stream_id on " << *socket
- << " h2req=" << (StreamUserData*)this;
- return butil::Status(EH2RUNOUTSTREAMS, "Fail to allocate stream_id");
- }
-
- _sctx->Init(ctx, id);
- // Insert the stream before encoding the headers and do not fail after
- // encoding: HPACK encoding adds the headers to the dynamic table of the
- // connection, and if the request were refused afterwards, the caller
- // would drop the HEADERS while the table kept the entries, so the
- // server would decode later requests on this connection with wrong
- // headers. The DATA that the windows allow now follows the HEADERS.
const butil::IOBuf& request_data = _cntl->request_attachment();
- butil::IOBuf data_frames;
- const butil::Status insert_status = ctx->TryToInsertClientStream(
- id, _sctx.get(), request_data, &data_frames);
- if (!insert_status.ok()) {
- return insert_status;
- }
- _stream_id = id;
- // TryToInsertClientStream transfers ownership of _sctx to ctx on success.
- _sctx.release();
-
- 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;
- const H2Settings remote_settings = ctx->remote_settings();
- if (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 (_cntl->has_http_request()) {
- const HttpHeader& h = _cntl->http_request();
- for (HttpHeader::HeaderIterator it = h.HeaderBegin();
- it != h.HeaderEnd(); ++it) {
- HPacker::Header header(it->first, it->second);
- hpacker.Encode(&appender, header, options);
+ {
+ std::unique_lock<butil::Mutex> stream_mu(ctx->_stream_mutex);
+ if (ctx->RunOutStreams()) {
+ // The RPC should be failed and retried.
+ // Note that the socket should not be SetFailed() which may affect
+ // other RPC successfully sent requests and waiting for responses.
+ LOG(WARNING) << "Fail to allocate new client stream,
_last_sent_stream_id="
+ << ctx->_last_sent_stream_id;
+ RPC_VLOG << "Fail to allocate stream_id on " << *socket
+ << " h2req=" << (StreamUserData*)this;
+ return butil::Status(EH2RUNOUTSTREAMS, "Fail to allocate
stream_id");
+ }
+ const int id = static_cast<int>(ctx->_last_sent_stream_id);
+ butil::IOBuf data_out;
+ const butil::Status st = ctx->TryToInsertClientStreamLocked(
+ id, _sctx.get(), request_data, &data_out);
+ if (!st.ok()) {
+ return st;
+ }
+
+ // Mutate connection state and HPACK dynamic table only after
+ // TryToInsertClientStreamLocked succeeds.
+ ctx->_last_sent_stream_id += 2;
+ if (ctx->_client_preface_pending) {
+ ctx->_client_preface_pending = false;
+ out->append(H2_CONNECTION_PREFACE_PREFIX,
+ H2_CONNECTION_PREFACE_PREFIX_SIZE);
+ char settingsbuf[FRAME_HEAD_SIZE + H2_SETTINGS_MAX_BYTE_SIZE +
+ FRAME_HEAD_SIZE + 4/*for WU*/];
+ const size_t nb = SerializeH2SettingsFrameAndWU(
+ ctx->_unack_local_settings, settingsbuf);
+ out->append(settingsbuf, nb);
+ }
+
+ 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 (_cntl->has_http_request()) {
+ const HttpHeader& h = _cntl->http_request();
+ for (HttpHeader::HeaderIterator it = h.HeaderBegin();
+ it != h.HeaderEnd(); ++it) {
+ HPacker::Header header(it->first, it->second);
+ hpacker.Encode(&appender, header, options);
+ }
}
+ butil::IOBuf frag;
+ appender.move_to(frag);
+ PackH2Headers(out, frag, id, ctx->_remote_settings.max_frame_size,
+ request_data.empty());
+ out->append(butil::IOBuf::Movable(data_out));
+ _stream_id = id;
+ _sctx.release();
}
- butil::IOBuf frag;
- appender.move_to(frag);
- PackH2Headers(out, frag, id, remote_settings.max_frame_size,
- request_data.empty());
- out->append(butil::IOBuf::Movable(data_frames));
const int64_t conn_wu = ctx->ReleaseDeferredWindowUpdate();
if (conn_wu > 0) {
char winbuf[FRAME_HEAD_SIZE + 4];
@@ -2019,57 +2183,152 @@ 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;
+ H2StreamContext* notify_failed_sctx = nullptr;
+ bthread_id_t deferred_notify_id = INVALID_BTHREAD_ID;
+ int notify_reg_error = 0;
+ {
+ 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 {
+ const int rc = bthread_id_create(
+ &deferred_notify_id, sctx,
H2StreamContext::OnSocketFailed);
+ if (rc == 0) {
+ CHECK_EQ(0, bthread_id_trylock(deferred_notify_id, NULL));
+ // 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);
+ } else {
+ notify_reg_error = rc;
+ notify_failed_sctx = ctx->RemoveStreamLocked(_stream_id);
+ }
+ }
+ }
+ }
+ if (deferred_notify_id != INVALID_BTHREAD_ID) {
+ socket->NotifyOnFailed(deferred_notify_id);
+ notify_reg_error = bthread_id_unlock(deferred_notify_id);
+ }
+ if (notify_failed_sctx != nullptr) {
+
ctx->DeferWindowUpdate(notify_failed_sctx->ReleaseDeferredWindowUpdate());
+ notify_failed_sctx->RemoveOneRefForStage2();
+ }
+ if (notify_reg_error != 0) {
+ LOG_EVERY_SECOND(WARNING)
+ << "Fail to register deferred HTTP/2 response notification on "
+ << *socket << ": " << berror(notify_reg_error)
+ << ", resetting stream_id=" << _stream_id;
+ char rstbuf[FRAME_HEAD_SIZE + 4];
+ SerializeFrameHead(rstbuf, 4, H2_FRAME_RST_STREAM, 0, _stream_id);
+ SaveUint32(rstbuf + FRAME_HEAD_SIZE, H2_INTERNAL_ERROR);
+ out->append(rstbuf, sizeof(rstbuf));
+ get_h2_bvars()->h2_stream_reset_count << 1;
+ }
+ if (overcrowded_sctx != nullptr) {
+
ctx->DeferWindowUpdate(overcrowded_sctx->ReleaseDeferredWindowUpdate());
+ overcrowded_sctx->RemoveOneRefForStage2();
+ LOG_EVERY_SECOND(WARNING)
+ << "Pending HTTP/2 response data on " << *socket
+ << " exceeds limit=" << GetH2MaxPendingDataBytes()
+ << ", resetting stream_id=" << _stream_id;
char rstbuf[FRAME_HEAD_SIZE + 4];
SerializeFrameHead(rstbuf, 4, H2_FRAME_RST_STREAM, 0, _stream_id);
- SaveUint32(rstbuf + FRAME_HEAD_SIZE, H2_FLOW_CONTROL_ERROR);
+ SaveUint32(rstbuf + FRAME_HEAD_SIZE, H2_ENHANCE_YOUR_CALM);
out->append(rstbuf, sizeof(rstbuf));
+ get_h2_bvars()->h2_stream_reset_count << 1;
return butil::Status::OK();
}
-
- 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 (finished_sctx != nullptr) {
+ ctx->DeferWindowUpdate(finished_sctx->ReleaseDeferredWindowUpdate());
+ finished_sctx->RemoveOneRefForStage2();
}
- 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);
-
- butil::IOBuf trailer_frag;
- if (_is_grpc) {
- HPacker::Header status_header("grpc-status",
- butil::string_printf("%d",
_grpc_status));
- hpacker.Encode(&appender, status_header, options);
- if (!_grpc_message.empty()) {
- HPacker::Header msg_header("grpc-message", _grpc_message);
- hpacker.Encode(&appender, msg_header, options);
- }
- appender.move_to(trailer_frag);
+ const int64_t conn_wu = ctx->ReleaseDeferredWindowUpdate();
+ if (conn_wu > 0) {
+ char winbuf[FRAME_HEAD_SIZE + 4];
+ SerializeFrameHead(winbuf, 4, H2_FRAME_WINDOW_UPDATE, 0, 0);
+ SaveUint32(winbuf + FRAME_HEAD_SIZE, conn_wu);
+ out->append(winbuf, sizeof(winbuf));
}
-
- PackH2Message(out, frag, trailer_frag, _data, _stream_id, ctx);
return butil::Status::OK();
}
diff --git a/src/brpc/policy/http2_rpc_protocol.h
b/src/brpc/policy/http2_rpc_protocol.h
index b82dd93f5..6c0ed8fe4 100644
--- a/src/brpc/policy/http2_rpc_protocol.h
+++ b/src/brpc/policy/http2_rpc_protocol.h
@@ -25,10 +25,7 @@
#include "brpc/details/hpack.h"
#include "brpc/stream_creator.h"
#include "brpc/controller.h"
-
-#ifndef NDEBUG
#include "bvar/bvar.h"
-#endif
namespace brpc {
namespace policy {
@@ -115,20 +112,22 @@ enum H2StreamState {
};
const char* H2StreamState2Str(H2StreamState);
-#ifndef NDEBUG
struct H2Bvars {
bvar::Adder<int> h2_unsent_request_count;
bvar::Adder<int> h2_stream_context_count;
+ bvar::Adder<int64_t> h2_stream_reset_count;
+ bvar::Adder<int64_t> h2_pending_data_bytes;
H2Bvars()
: h2_unsent_request_count("h2_unsent_request_count")
- , h2_stream_context_count("h2_stream_context_count") {
+ , h2_stream_context_count("h2_stream_context_count")
+ , h2_stream_reset_count("h2_stream_reset_count")
+ , h2_pending_data_bytes("h2_pending_data_bytes") {
}
};
inline H2Bvars* get_h2_bvars() {
return butil::get_leaky_singleton<H2Bvars>();
}
-#endif
class H2UnsentRequest : public SocketMessage, public StreamUserData {
friend void PackH2Request(butil::IOBuf*, SocketMessage**,
@@ -244,6 +243,8 @@ public:
size_t parsed_length() const { return this->_parsed_length; }
int stream_id() const { return _stream_id; }
+ bool rst_received() const { return _rst_received; }
+ H2Error rst_error() const { return _rst_error; }
int64_t ReleaseDeferredWindowUpdate() {
if (_deferred_window_update.load(butil::memory_order_relaxed) == 0) {
@@ -252,6 +253,39 @@ public:
return _deferred_window_update.exchange(0,
butil::memory_order_relaxed);
}
+ static int OnSocketFailed(bthread_id_t id, void* data, int error_code);
+
+ void HoldDeferredResponseSocket(Socket* socket, bthread_id_t notify_id) {
+ CHECK_EQ(INVALID_BTHREAD_ID.value,
+ _deferred_response_id.load(butil::memory_order_relaxed));
+ butil::intrusive_ptr<H2StreamContext>(this).detach();
+ socket->AddIdleFailRef();
+ socket->ReAddress(&_deferred_response_sock);
+ _deferred_response_id.store(notify_id.value,
butil::memory_order_release);
+ }
+
+ void ReleaseDeferredResponseSocket() {
+ const uint64_t id_val = _deferred_response_id.exchange(
+ INVALID_BTHREAD_ID.value, butil::memory_order_acq_rel);
+ if (id_val != INVALID_BTHREAD_ID.value) {
+ const bthread_id_t id = {id_val};
+ bthread_id_error(id, 0);
+ }
+ }
+
+ void RemoveOneRefForStage2() {
+ ReleaseDeferredResponseSocket();
+ HttpContext::RemoveOneRefForStage2();
+ }
+
+ // @InputMessageBase
+ void DestroyImpl() override {
+ if (!read_body_progressively()) {
+ body().clear();
+ }
+ HttpContext::DestroyImpl();
+ }
+
#if defined(BRPC_H2_STREAM_STATE)
H2StreamState state() const { return _state; }
void SetState(H2StreamState state);
@@ -264,6 +298,10 @@ friend class H2Context;
#endif
int _stream_id;
bool _stream_ended;
+ bool _remote_ended;
+ bool _response_committed;
+ bool _rst_received;
+ H2Error _rst_error;
butil::atomic<int64_t> _remote_window_left;
butil::atomic<int64_t> _deferred_window_update;
uint64_t _correlation_id;
@@ -284,10 +322,18 @@ friend class H2Context;
// the connection is closed." Hence the rejection is remembered here and
// turned into a RST_STREAM once END_HEADERS is reached.
H2Error _rejected_error;
+ butil::atomic<uint64_t> _deferred_response_id;
+ SocketUniquePtr _deferred_response_sock;
butil::IOBuf _remaining_header_fragment;
- // Request body which cannot be sent yet due to remote flow control.
- // Accessed under H2Context::_stream_mutex.
+ // Request or response body which cannot be sent yet due to remote flow
control.
+ // Accessed under H2Context::_stream_mutex. Bounded per connection by
+ // -h2_max_pending_data_bytes (or -socket_max_unwritten_bytes when 0);
+ // idle connections with stalled peers that withhold WINDOW_UPDATE can be
+ // reclaimed when ServerOptions::idle_timeout_sec is enabled.
butil::IOBuf _pending_data;
+ // Encoded gRPC trailer headers waiting for _pending_data to be flushed.
+ // Accessed under H2Context::_stream_mutex.
+ butil::IOBuf _pending_trailer;
};
StreamCreator* get_h2_global_stream_creator();
@@ -386,19 +432,29 @@ friend void InitFrameHandlers();
H2ParseResult OnWindowUpdate(butil::IOBufBytesIterator&, const
H2FrameHead&);
H2ParseResult OnContinuation(butil::IOBufBytesIterator&, const
H2FrameHead&);
+ H2StreamContext* RemoveStreamLocked(int stream_id);
H2StreamContext* RemoveStreamAndDeferWU(int stream_id);
void RemoveGoAwayStreams(int goaway_stream_id,
std::vector<H2StreamContext*>* out_streams);
H2StreamContext* FindStream(int stream_id);
+ butil::intrusive_ptr<H2StreamContext> FindOpenStream(int stream_id);
+ butil::intrusive_ptr<H2StreamContext> FindHeaderStream(int stream_id);
// Atomically checks stream and pending-DATA limits, inserts the client
// stream, and appends DATA allowed by the current remote windows. On
// success, ownership of sctx is transferred to this context. On failure,
// no stream, window, or pending-DATA state is changed.
butil::Status TryToInsertClientStream(
int stream_id, H2StreamContext*, const butil::IOBuf&, butil::IOBuf*);
+ butil::Status TryToInsertClientStreamLocked(
+ int stream_id, H2StreamContext*, const butil::IOBuf&, butil::IOBuf*);
void AppendPendingDataLocked(H2StreamContext*, butil::IOBuf*);
- void ClearPendingData(int stream_id);
+ // Clears any unsent _pending_data / _pending_trailer for stream_id.
+ // Returns true if the stream still had unsent data/trailers (i.e.
END_STREAM
+ // had not been sent yet).
+ bool ClearPendingData(int stream_id);
bool FlushPendingData(int stream_id);
+ void AddPendingDataSize(size_t n);
+ void SubPendingDataSize(size_t n);
// True if the connection is established by client, otherwise it's
// accepted by server.
@@ -410,6 +466,7 @@ friend void InitFrameHandlers();
int _goaway_stream_id;
H2Settings _remote_settings;
bool _remote_settings_received;
+ bool _client_preface_pending;
H2Settings _local_settings;
H2Settings _unack_local_settings;
HPacker _hpacker;
@@ -418,9 +475,14 @@ friend void InitFrameHandlers();
typedef butil::FlatMap<int, H2StreamContext*> StreamMap;
mutable butil::Mutex _stream_mutex;
StreamMap _pending_streams;
+ // Relaxed atomic snapshot of _pending_streams.size(), updated under
+ // _stream_mutex and read lock-free by VolatilePendingStreamSize().
+ butil::atomic<size_t> _volatile_pending_stream_size;
// Total bytes retained in H2StreamContext::_pending_data on this
- // connection. Accessed under _stream_mutex.
- size_t _pending_data_size;
+ // connection. Modified under _stream_mutex; read lock-free via relaxed
+ // atomic load in PendingDataOvercrowded().
+ butil::atomic<size_t> _pending_data_size;
+ int _last_flushed_stream_id;
butil::atomic<int64_t> _deferred_window_update;
};
diff --git a/src/brpc/policy/http_rpc_protocol.cpp
b/src/brpc/policy/http_rpc_protocol.cpp
index e4bac8abd..cb7f8d8c6 100644
--- a/src/brpc/policy/http_rpc_protocol.cpp
+++ b/src/brpc/policy/http_rpc_protocol.cpp
@@ -468,12 +468,27 @@ void ProcessHttpResponse(InputMessageBase* msg) {
res_body.copy_to(
&body_str, std::min((int)res_body.size(),
FLAGS_http_max_error_length));
- cntl->SetFailed(EHTTP, "HTTP/%d.%d %d %s: %.*s",
- res_header->major_version(),
- res_header->minor_version(),
- static_cast<int>(res_header->status_code()),
- res_header->reason_phrase(),
- (int)body_str.size(), body_str.c_str());
+ const H2StreamContext* h2_sctx =
+ is_http2 ? static_cast<const
H2StreamContext*>(imsg_guard.get()) : nullptr;
+ if (h2_sctx != nullptr && h2_sctx->rst_received()) {
+ cntl->SetFailed(EHTTP,
+ "HTTP/%d.%d %d %s (stream_id=%d terminated
by RST_STREAM: %s[0x%x]): %.*s",
+ res_header->major_version(),
+ res_header->minor_version(),
+
static_cast<int>(res_header->status_code()),
+ res_header->reason_phrase(),
+ h2_sctx->stream_id(),
+ H2ErrorToString(h2_sctx->rst_error()),
+ static_cast<unsigned
int>(h2_sctx->rst_error()),
+ (int)body_str.size(), body_str.c_str());
+ } else {
+ cntl->SetFailed(EHTTP, "HTTP/%d.%d %d %s: %.*s",
+ res_header->major_version(),
+ res_header->minor_version(),
+
static_cast<int>(res_header->status_code()),
+ res_header->reason_phrase(),
+ (int)body_str.size(), body_str.c_str());
+ }
} else if (cntl->response() != nullptr &&
cntl->response()->GetDescriptor()->field_count() != 0) {
cntl->SetFailed(ERESPONSE, "A protobuf response can't be
parsed"
@@ -492,6 +507,17 @@ void ProcessHttpResponse(InputMessageBase* msg) {
res_header->minor_version(),
static_cast<int>(res_header->status_code()),
res_header->reason_phrase());
+ if (is_http2) {
+ const H2StreamContext* h2_sctx =
+ static_cast<const H2StreamContext*>(imsg_guard.get());
+ if (h2_sctx->rst_received()) {
+ butil::string_appendf(
+ &err, " (stream_id=%d terminated by RST_STREAM:
%s[0x%x])",
+ h2_sctx->stream_id(),
+ H2ErrorToString(h2_sctx->rst_error()),
+ static_cast<unsigned int>(h2_sctx->rst_error()));
+ }
+ }
if (!res_body.empty()) {
// Use content as error text if it's present. Notice that
// content may be binary data, so the size limit is a must.
@@ -868,6 +894,12 @@ HttpResponseSender::~HttpResponseSender() {
const google::protobuf::Message* res = nullptr != _messages ?
_messages->Response() : nullptr;
if (cntl->IsCloseConnection()) {
+ if (cntl->http_request().is_http2()) {
+ H2Context* ctx =
static_cast<H2Context*>(socket->parsing_context());
+ if (ctx != NULL) {
+ ctx->AddAbandonedStream(_h2_stream_id);
+ }
+ }
socket->SetFailed();
return;
}
@@ -1066,6 +1098,12 @@ HttpResponseSender::~HttpResponseSender() {
}
if (rc != 0) {
+ if (is_http2) {
+ H2Context* ctx =
static_cast<H2Context*>(socket->parsing_context());
+ if (ctx != NULL) {
+ ctx->AddAbandonedStream(_h2_stream_id);
+ }
+ }
// EPIPE is common in pooled connections + backup requests.
const int errcode = errno;
PLOG_IF(WARNING, errcode != EPIPE) << "Fail to write into " << *socket;
@@ -1077,6 +1115,10 @@ HttpResponseSender::~HttpResponseSender() {
bthread_id_join(response_id);
// Do not care about the result of background writing.
// TODO: this is not sent
+ // Note: Under HTTP/2 with notify_on_success=true,
H2UnsentResponse::AppendAndDestroySelf
+ // returns OK once HEADERS and the initial window of DATA frames are
queued, while
+ // remaining DATA / Trailer frames may be deferred until WINDOW_UPDATE
arrives.
+ // Thus args.sent_us records when the initial response frames are
written.
span->set_sent_us(args.sent_us);
}
}
@@ -1775,6 +1817,8 @@ void ProcessHttpRequest(InputMessageBase *msg) {
MakeRawHttpRequest(&sample->request, &req_header, ep,
&req_body);
sample->submit(start_parse_us);
}
+ } else {
+ req_body.clear();
}
} else {
if (imsg_guard->read_body_progressively()) {
diff --git a/src/brpc/socket.cpp b/src/brpc/socket.cpp
index 2985a35b6..65c00f58e 100644
--- a/src/brpc/socket.cpp
+++ b/src/brpc/socket.cpp
@@ -477,6 +477,7 @@ Socket::Socket(Forbidden f)
, _connection_type_for_progressive_read(CONNECTION_TYPE_UNKNOWN)
, _controller_released_socket(false)
, _fail_me_at_server_stop(false)
+ , _idle_fail_ref_count(0)
, _logoff_flag(false)
, _error_code(0)
, _pipeline_q(nullptr)
@@ -774,6 +775,7 @@ int Socket::OnCreated(const SocketOptions& options) {
_controller_released_socket.store(false, butil::memory_order_relaxed);
// Maybe non-zero for RTMP connections.
_fail_me_at_server_stop = false;
+ _idle_fail_ref_count.store(0, butil::memory_order_relaxed);
_logoff_flag.store(false, butil::memory_order_relaxed);
_error_code = 0;
_agent_socket_id.store(INVALID_SOCKET_ID, butil::memory_order_relaxed);
@@ -1113,10 +1115,11 @@ int Socket::ReleaseReferenceIfIdle(int idle_seconds) {
LOG_IF(WARNING, FLAGS_log_idle_connection_close)
<< "Close " << *this << " due to no data transmission for "
<< idle_seconds << " seconds";
- if (shall_fail_me_at_server_stop()) {
- // sockets for streaming purposes (say RTMP) are probably referenced
- // by many places, ReleaseAdditionalReference() cannot notify other
- // places to release refs, SetFailed() is a must.
+ if (shall_fail_me_at_server_stop() || shall_fail_me_at_idle_timeout()) {
+ // sockets for streaming purposes (say RTMP) or holding deferred
+ // responses are probably referenced by other places,
+ // ReleaseAdditionalReference() cannot notify other places to release
+ // refs, SetFailed() is a must.
return SetFailed(EUNUSED, "No data transmission for %d seconds",
idle_seconds);
}
diff --git a/src/brpc/socket.h b/src/brpc/socket.h
index 8a1d37229..b947202c8 100644
--- a/src/brpc/socket.h
+++ b/src/brpc/socket.h
@@ -691,6 +691,21 @@ public:
void fail_me_at_server_stop() { _fail_me_at_server_stop = true; }
bool shall_fail_me_at_server_stop() const { return
_fail_me_at_server_stop; }
+ // Track references held by deferred responses waiting for peer activity
+ // (such as HTTP/2 WINDOW_UPDATE). Unlike fail_me_at_server_stop(), these
+ // references are allowed to drain gracefully on Server.Stop(), but must
+ // trigger SetFailed() when the socket becomes idle so NotifyOnFailed()
+ // callbacks can clean up buffered state and release the socket.
+ void AddIdleFailRef() {
+ _idle_fail_ref_count.fetch_add(1, butil::memory_order_relaxed);
+ }
+ void ReleaseIdleFailRef() {
+ _idle_fail_ref_count.fetch_sub(1, butil::memory_order_relaxed);
+ }
+ bool shall_fail_me_at_idle_timeout() const {
+ return _idle_fail_ref_count.load(butil::memory_order_relaxed) > 0;
+ }
+
// Tag the socket so that the response coming back from socket will be
// parsed progressively. For example: in HTTP, the RPC may end w/o reading
// the body part fully.
@@ -1024,6 +1039,7 @@ private:
butil::atomic<bool> _controller_released_socket;
bool _fail_me_at_server_stop;
+ butil::atomic<int32_t> _idle_fail_ref_count;
// Set by SetLogOff
butil::atomic<bool> _logoff_flag;
diff --git a/test/brpc_grpc_protocol_unittest.cpp
b/test/brpc_grpc_protocol_unittest.cpp
index 41107b6e5..a71cd5e67 100644
--- a/test/brpc_grpc_protocol_unittest.cpp
+++ b/test/brpc_grpc_protocol_unittest.cpp
@@ -22,9 +22,17 @@
#include "brpc/server.h"
#include "brpc/channel.h"
#include "brpc/grpc.h"
+#include "bthread/bthread.h"
#include "butil/time.h"
#include "grpc.pb.h"
+namespace brpc {
+namespace policy {
+DECLARE_int32(h2_client_stream_window_size);
+DECLARE_int32(h2_client_connection_window_size);
+} // namespace policy
+} // namespace brpc
+
int main(int argc, char* argv[]) {
testing::InitGoogleTest(&argc, argv);
GFLAGS_NAMESPACE::ParseCommandLineFlags(&argc, &argv, true);
@@ -94,7 +102,18 @@ public:
::test::GrpcResponse* res,
::google::protobuf::Closure* done) override {
brpc::ClosureGuard done_guard(done);
- res->set_message(req->message());
+ const std::string& msg = req->message();
+ const std::string prefix = "resp_size:";
+ if (msg.compare(0, prefix.size(), prefix) == 0) {
+ size_t n = static_cast<size_t>(strtoull(msg.c_str() +
prefix.size(), nullptr, 10));
+ char fill = msg.back();
+ if (fill >= '0' && fill <= '9') {
+ fill = 'r';
+ }
+ res->set_message(std::string(n, fill));
+ } else {
+ res->set_message(msg);
+ }
}
};
@@ -329,4 +348,108 @@ TEST(GrpcProtocol,
client_sends_large_request_with_small_remote_window) {
EXPECT_EQ(request.message(), response.message());
}
+TEST(GrpcProtocol, server_sends_large_response_with_small_remote_window) {
+ GFLAGS_NAMESPACE::FlagSaver flag_saver;
+ // Simulate a gRPC client with a tiny stream window (256 bytes) and
+ // RFC 7540 default 64 KiB connection window receiving a 512 KiB response
+ // to a tiny request (like GetClusterOverallStatus).
+ brpc::policy::FLAGS_h2_client_stream_window_size = 256;
+ brpc::policy::FLAGS_h2_client_connection_window_size = 65535;
+
+ WindowGrpcService service;
+ brpc::Server server;
+ ASSERT_EQ(0, server.AddService(&service, brpc::SERVER_DOESNT_OWN_SERVICE));
+ ASSERT_EQ(0, server.Start("127.0.0.1:0", nullptr));
+
+ brpc::Channel channel;
+ brpc::ChannelOptions channel_options;
+ channel_options.protocol = g_protocol;
+ channel_options.timeout_ms = 10000;
+ ASSERT_EQ(0, channel.Init(server.listen_address(), &channel_options));
+ test::GrpcService_Stub stub(&channel);
+
+ // Warmup RPC so Server receives Client's SETTINGS (stream_window_size=256)
+ {
+ test::GrpcRequest request;
+ test::GrpcResponse response;
+ brpc::Controller cntl;
+ request.set_message("warmup");
+ request.set_gzip(false);
+ request.set_return_error(false);
+ stub.Method(&cntl, &request, &response, nullptr);
+ ASSERT_FALSE(cntl.Failed()) << cntl.ErrorText();
+ ASSERT_EQ("warmup", response.message());
+ }
+
+ // Request a 512 KiB response with a tiny request
+ const size_t resp_size = 512 * 1024;
+ test::GrpcRequest request;
+ test::GrpcResponse response;
+ brpc::Controller cntl;
+ request.set_message(butil::string_printf("resp_size:%zu:A", resp_size));
+ request.set_gzip(false);
+ request.set_return_error(false);
+ stub.Method(&cntl, &request, &response, nullptr);
+ EXPECT_FALSE(cntl.Failed()) << cntl.ErrorText();
+ EXPECT_EQ(std::string(resp_size, 'A'), response.message());
+}
+
+struct ConcurrentGrpcCallArg {
+ test::GrpcService_Stub* stub;
+ size_t resp_size;
+ char fill;
+ bool ok;
+};
+
+static void* RunConcurrentGrpcCall(void* void_arg) {
+ ConcurrentGrpcCallArg* arg = static_cast<ConcurrentGrpcCallArg*>(void_arg);
+ test::GrpcRequest request;
+ test::GrpcResponse response;
+ brpc::Controller cntl;
+ request.set_message(butil::string_printf("resp_size:%zu:%c",
arg->resp_size, arg->fill));
+ request.set_gzip(false);
+ request.set_return_error(false);
+ arg->stub->Method(&cntl, &request, &response, nullptr);
+ arg->ok = !cntl.Failed() && (response.message() ==
std::string(arg->resp_size, arg->fill));
+ if (cntl.Failed()) {
+ LOG(ERROR) << "Concurrent gRPC failed: " << cntl.ErrorText();
+ }
+ return nullptr;
+}
+
+TEST(GrpcProtocol, concurrent_large_responses_with_small_remote_window) {
+ GFLAGS_NAMESPACE::FlagSaver flag_saver;
+ // 8 concurrent streams each returning 128 KiB (1 MiB total) over a single
+ // HTTP/2 connection with 4 KiB stream window and 64 KiB connection window.
+ brpc::policy::FLAGS_h2_client_stream_window_size = 4096;
+ brpc::policy::FLAGS_h2_client_connection_window_size = 65535;
+
+ WindowGrpcService service;
+ brpc::Server server;
+ ASSERT_EQ(0, server.AddService(&service, brpc::SERVER_DOESNT_OWN_SERVICE));
+ ASSERT_EQ(0, server.Start("127.0.0.1:0", nullptr));
+
+ brpc::Channel channel;
+ brpc::ChannelOptions channel_options;
+ channel_options.protocol = g_protocol;
+ channel_options.timeout_ms = 15000;
+ ASSERT_EQ(0, channel.Init(server.listen_address(), &channel_options));
+ test::GrpcService_Stub stub(&channel);
+
+ const int kNumStreams = 8;
+ bthread_t tids[kNumStreams];
+ ConcurrentGrpcCallArg args[kNumStreams];
+ for (int i = 0; i < kNumStreams; ++i) {
+ args[i].stub = &stub;
+ args[i].resp_size = 128 * 1024;
+ args[i].fill = static_cast<char>('A' + i);
+ args[i].ok = false;
+ ASSERT_EQ(0, bthread_start_background(&tids[i], nullptr,
RunConcurrentGrpcCall, &args[i]));
+ }
+ for (int i = 0; i < kNumStreams; ++i) {
+ bthread_join(tids[i], nullptr);
+ EXPECT_TRUE(args[i].ok) << "Stream " << i << " failed";
+ }
+}
+
} // namespace
diff --git a/test/brpc_h2_unsent_message_unittest.cpp
b/test/brpc_h2_unsent_message_unittest.cpp
index 0d7b39c3d..35d0f7c5f 100644
--- a/test/brpc_h2_unsent_message_unittest.cpp
+++ b/test/brpc_h2_unsent_message_unittest.cpp
@@ -25,6 +25,7 @@
#include "bthread/bthread.h"
#include "butil/atomicops.h"
#include "brpc/details/hpack.h"
+#include "brpc/server.h"
#include "brpc/policy/http_rpc_protocol.h"
#include "brpc/policy/http2_rpc_protocol.h"
#include "gperftools_helper.h"
@@ -35,11 +36,15 @@ DECLARE_int64(socket_max_unwritten_bytes);
int main(int argc, char* argv[]) {
testing::InitGoogleTest(&argc, argv);
+ brpc::get_or_new_client_side_messenger();
return RUN_ALL_TESTS();
}
namespace {
+const uint8_t H2_FLAGS_END_STREAM = 0x01;
+const uint8_t H2_FLAGS_END_HEADERS = 0x04;
+
brpc::policy::H2FrameHead PopFrame(butil::IOBuf* buf, std::string* payload) {
char head[brpc::policy::FRAME_HEAD_SIZE];
CHECK_EQ(sizeof(head), buf->cutn(head, sizeof(head)));
@@ -140,7 +145,7 @@ TEST(H2UnsentMessage, split_request_data_by_remote_window) {
EXPECT_EQ("e", payload);
EXPECT_TRUE(out.empty());
EXPECT_EQ(5u, sctx->_pending_data.size());
- EXPECT_EQ(5u, ctx->_pending_data_size);
+ EXPECT_EQ(5u, ctx->_pending_data_size.load());
EXPECT_EQ(1, ctx->_remote_window_left);
EXPECT_EQ(0, sctx->_remote_window_left);
@@ -155,7 +160,7 @@ TEST(H2UnsentMessage, split_request_data_by_remote_window) {
EXPECT_EQ(0, frame.flags & 0x1);
EXPECT_EQ("fgh", payload);
EXPECT_TRUE(out.empty());
- EXPECT_EQ(2u, ctx->_pending_data_size);
+ EXPECT_EQ(2u, ctx->_pending_data_size.load());
ctx->_remote_window_left.fetch_add(2, butil::memory_order_relaxed);
sctx->_remote_window_left.fetch_add(2, butil::memory_order_relaxed);
@@ -169,7 +174,7 @@ TEST(H2UnsentMessage, split_request_data_by_remote_window) {
EXPECT_EQ("ij", payload);
EXPECT_TRUE(out.empty());
EXPECT_TRUE(sctx->_pending_data.empty());
- EXPECT_EQ(0u, ctx->_pending_data_size);
+ EXPECT_EQ(0u, ctx->_pending_data_size.load());
}
TEST(H2UnsentMessage, request_does_not_fail_when_body_exceeds_window) {
@@ -216,7 +221,7 @@ TEST(H2UnsentMessage,
request_does_not_fail_when_body_exceeds_window) {
brpc::policy::H2StreamContext* sctx = ctx->FindStream(1);
ASSERT_TRUE(sctx != nullptr);
EXPECT_EQ(7u, sctx->_pending_data.size());
- EXPECT_EQ(7u, ctx->_pending_data_size);
+ EXPECT_EQ(7u, ctx->_pending_data_size.load());
EXPECT_EQ(0, ctx->_remote_window_left);
EXPECT_EQ(0, sctx->_remote_window_left);
}
@@ -237,16 +242,43 @@ TEST(H2UnsentMessage,
invalid_window_update_does_not_change_window) {
ctx->_remote_window_left = max_window_size;
const char increment[] = {0, 0, 0, 1};
- butil::IOBuf payload;
- payload.append(increment, sizeof(increment));
- butil::IOBufBytesIterator it(payload);
- const brpc::policy::H2FrameHead frame = {
- 4, brpc::policy::H2_FRAME_WINDOW_UPDATE, 0, 0};
- const brpc::policy::H2ParseResult result = ctx->OnWindowUpdate(it, frame);
-
- EXPECT_EQ(brpc::H2_FLOW_CONTROL_ERROR, result.error());
- EXPECT_EQ(max_window_size,
- ctx->_remote_window_left.load(butil::memory_order_relaxed));
+ {
+ butil::IOBuf payload;
+ payload.append(increment, sizeof(increment));
+ butil::IOBufBytesIterator it(payload);
+ const brpc::policy::H2FrameHead frame = {
+ 4, brpc::policy::H2_FRAME_WINDOW_UPDATE, 0, 0};
+ const brpc::policy::H2ParseResult result = ctx->OnWindowUpdate(it,
frame);
+
+ EXPECT_EQ(brpc::H2_FLOW_CONTROL_ERROR, result.error());
+ EXPECT_EQ(0, result.stream_id());
+ EXPECT_EQ(max_window_size,
+ ctx->_remote_window_left.load(butil::memory_order_relaxed));
+ }
+
+ // Stream-level WINDOW_UPDATE overflow must return a stream error
(stream_id = 1)
+ // per RFC 7540 Section 6.9.1 rather than a connection error.
+ brpc::policy::H2StreamContext* sctx =
+ new brpc::policy::H2StreamContext(false);
+ sctx->Init(ctx, 1);
+ ASSERT_EQ(0, ctx->TryToInsertStream(1, sctx));
+ sctx->_remote_window_left = max_window_size;
+ {
+ butil::IOBuf payload;
+ payload.append(increment, sizeof(increment));
+ butil::IOBufBytesIterator it(payload);
+ const brpc::policy::H2FrameHead frame = {
+ 4, brpc::policy::H2_FRAME_WINDOW_UPDATE, 0, 1};
+ const brpc::policy::H2ParseResult result = ctx->OnWindowUpdate(it,
frame);
+
+ EXPECT_EQ(brpc::H2_FLOW_CONTROL_ERROR, result.error());
+ EXPECT_EQ(1, result.stream_id());
+ EXPECT_EQ(max_window_size,
+ sctx->_remote_window_left.load(butil::memory_order_relaxed));
+ }
+ brpc::policy::H2StreamContext* removed_sctx =
ctx->RemoveStreamAndDeferWU(1);
+ ASSERT_EQ(sctx, removed_sctx);
+ removed_sctx->RemoveOneRefForStage2();
}
TEST(H2UnsentMessage, clear_pending_data_releases_overcrowded_buffer) {
@@ -276,7 +308,7 @@ TEST(H2UnsentMessage,
clear_pending_data_releases_overcrowded_buffer) {
ctx->TryToInsertClientStream(1, sctx.get(), body, &out).ok());
brpc::policy::H2StreamContext* inserted_sctx = sctx.release();
ASSERT_TRUE(out.empty());
- ASSERT_EQ(body.size(), ctx->_pending_data_size);
+ ASSERT_EQ(body.size(), ctx->_pending_data_size.load());
EXPECT_TRUE(ctx->PendingDataOvercrowded());
std::unique_ptr<brpc::policy::H2StreamContext> rejected_sctx(
@@ -288,13 +320,14 @@ TEST(H2UnsentMessage,
clear_pending_data_releases_overcrowded_buffer) {
3, rejected_sctx.get(), rejected_body, &out);
EXPECT_EQ(brpc::EOVERCROWDED, rejected.error_code());
EXPECT_EQ(nullptr, ctx->FindStream(3));
- EXPECT_EQ(body.size(), ctx->_pending_data_size);
+ EXPECT_EQ(body.size(), ctx->_pending_data_size.load());
- ctx->ClearPendingData(1);
+ EXPECT_TRUE(ctx->ClearPendingData(1));
+ EXPECT_FALSE(ctx->ClearPendingData(1));
EXPECT_FALSE(ctx->PendingDataOvercrowded());
EXPECT_TRUE(inserted_sctx->_pending_data.empty());
- EXPECT_EQ(0u, ctx->_pending_data_size);
+ EXPECT_EQ(0u, ctx->_pending_data_size.load());
}
TEST(H2UnsentMessage,
append_and_destroy_self_keeps_hpack_state_on_refused_insert) {
@@ -352,7 +385,7 @@ TEST(H2UnsentMessage, request_throughput) {
brpc::Controller cntl;
butil::IOBuf request_buf;
cntl.http_request().uri() = "0.0.0.0:8010/HttpService/Echo";
- brpc::policy::SerializeHttpRequest(&request_buf, &cntl, nullptr);
+ brpc::policy::SerializeHttpRequest(&request_buf, &cntl, NULL);
brpc::SocketId id;
brpc::SocketUniquePtr h2_client_sock;
@@ -362,7 +395,7 @@ TEST(H2UnsentMessage, request_throughput) {
EXPECT_EQ(0, brpc::Socket::Address(id, &h2_client_sock));
brpc::policy::H2Context* ctx =
- new brpc::policy::H2Context(h2_client_sock.get(), nullptr);
+ new brpc::policy::H2Context(h2_client_sock.get(), NULL);
CHECK_EQ(ctx->Init(), 0);
h2_client_sock->initialize_parsing_context(&ctx);
ctx->_last_sent_stream_id = 0;
@@ -386,6 +419,30 @@ TEST(H2UnsentMessage, request_throughput) {
<< dummy_buf.size() * 1000000L / elapsed << "/s";
// calc H2UnsentResponse throughput
+ brpc::Server server;
+ brpc::SocketId srv_id;
+ brpc::SocketUniquePtr h2_server_sock;
+ brpc::SocketOptions h2_server_options;
+ EXPECT_EQ(0, brpc::Socket::Create(h2_server_options, &srv_id));
+ EXPECT_EQ(0, brpc::Socket::Address(srv_id, &h2_server_sock));
+
+ brpc::policy::H2Context* srv_ctx =
+ new brpc::policy::H2Context(h2_server_sock.get(), &server);
+ CHECK_EQ(srv_ctx->Init(), 0);
+ h2_server_sock->initialize_parsing_context(&srv_ctx);
+ srv_ctx->_unack_local_settings.max_concurrent_streams =
+ std::numeric_limits<uint32_t>::max();
+ srv_ctx->_remote_settings.stream_window_size =
brpc::H2Settings::MAX_WINDOW_SIZE;
+ srv_ctx->_remote_window_left = brpc::H2Settings::MAX_WINDOW_SIZE;
+ for (int i = 0; i < ntotal; ++i) {
+ const int stream_id = 2 * i + 1;
+ brpc::policy::H2StreamContext* sctx =
+ new brpc::policy::H2StreamContext(false);
+ sctx->Init(srv_ctx, stream_id);
+ sctx->_remote_ended = true;
+ CHECK_EQ(0, srv_ctx->TryToInsertStream(stream_id, sctx));
+ }
+
dummy_buf.clear();
start_us = butil::cpuwide_time_us();
for (int i = 0; i < ntotal; ++i) {
@@ -393,8 +450,9 @@ TEST(H2UnsentMessage, request_throughput) {
// cntl.response_attachment()
cntl.http_response().set_content_type("text/plain");
cntl.response_attachment().append("0123456789abcedef");
- brpc::policy::H2UnsentResponse* res =
brpc::policy::H2UnsentResponse::New(&cntl, 0, false);
- res->AppendAndDestroySelf(&dummy_buf, h2_client_sock.get());
+ brpc::policy::H2UnsentResponse* res =
+ brpc::policy::H2UnsentResponse::New(&cntl, 2 * i + 1, false);
+ res->AppendAndDestroySelf(&dummy_buf, h2_server_sock.get());
}
end_us = butil::cpuwide_time_us();
elapsed = end_us - start_us;
@@ -402,3 +460,722 @@ TEST(H2UnsentMessage, request_throughput) {
<< (ntotal * 1000000L) / elapsed << "/s, data throughput="
<< dummy_buf.size() * 1000000L / elapsed << "/s";
}
+
+TEST(H2UnsentMessage,
split_response_data_and_defer_grpc_trailer_by_remote_window) {
+ brpc::Server server;
+ brpc::SocketId id;
+ brpc::SocketUniquePtr sock;
+ brpc::SocketOptions options;
+ ASSERT_EQ(0, brpc::Socket::Create(options, &id));
+ ASSERT_EQ(0, brpc::Socket::Address(id, &sock));
+ ASSERT_TRUE( !sock->CreatedByConnect());
+
+ brpc::policy::H2Context* ctx =
+ new brpc::policy::H2Context(sock.get(), &server);
+ ASSERT_EQ(0, ctx->Init());
+ sock->initialize_parsing_context(&ctx);
+ ctx->_remote_settings.max_frame_size = 4;
+ ctx->_remote_settings.stream_window_size = 5;
+ ctx->_remote_window_left = 6;
+
+ brpc::policy::H2StreamContext* sctx =
+ new brpc::policy::H2StreamContext(false);
+ sctx->Init(ctx, 1);
+ ASSERT_EQ(0, ctx->TryToInsertStream(1, sctx));
+ // Simulate OnEndStream() + Stage 2 ProcessHttpRequest consuming its ref
+ sctx->_remote_ended = true;
+
+ brpc::Controller cntl;
+ cntl.http_response().set_content_type("application/grpc");
+ cntl.response_attachment().append("abcdefghij");
+ brpc::policy::H2UnsentResponse* response =
+ brpc::policy::H2UnsentResponse::New(&cntl, 1, true /* is_grpc */);
+ ASSERT_TRUE(response != nullptr);
+
+ butil::IOBuf out;
+ ASSERT_TRUE(response->AppendAndDestroySelf(&out, sock.get()).ok());
+
+ std::string payload;
+ // Frame 1+: initial HEADERS (+ CONTINUATION if max_frame_size=4 splits it)
+ brpc::policy::H2FrameHead frame = PopFrame(&out, &payload);
+ EXPECT_EQ(brpc::policy::H2_FRAME_HEADERS, frame.type);
+ EXPECT_EQ(1, frame.stream_id);
+ while ((frame.flags & H2_FLAGS_END_HEADERS) == 0) {
+ frame = PopFrame(&out, &payload);
+ EXPECT_EQ(brpc::policy::H2_FRAME_CONTINUATION, frame.type);
+ EXPECT_EQ(1, frame.stream_id);
+ }
+ EXPECT_EQ(0, frame.flags & H2_FLAGS_END_STREAM);
+
+ // Next Frame: DATA "abcd" (4 bytes)
+ frame = PopFrame(&out, &payload);
+ EXPECT_EQ(brpc::policy::H2_FRAME_DATA, frame.type);
+ EXPECT_EQ(4u, frame.payload_size);
+ EXPECT_EQ(0, frame.flags & H2_FLAGS_END_STREAM);
+ EXPECT_EQ("abcd", payload);
+
+ // Next Frame: DATA "e" (1 byte, limited by stream window 5)
+ frame = PopFrame(&out, &payload);
+ EXPECT_EQ(brpc::policy::H2_FRAME_DATA, frame.type);
+ EXPECT_EQ(1u, frame.payload_size);
+ EXPECT_EQ(0, frame.flags & H2_FLAGS_END_STREAM);
+ EXPECT_EQ("e", payload);
+
+ // Trailer must be deferred because 5 bytes of DATA are still pending
+ EXPECT_TRUE(out.empty());
+ EXPECT_EQ(5u, sctx->_pending_data.size());
+ EXPECT_EQ(5u, ctx->_pending_data_size.load());
+ EXPECT_FALSE(sctx->_pending_trailer.empty());
+ EXPECT_EQ(1, ctx->_remote_window_left);
+ EXPECT_EQ(0, sctx->_remote_window_left);
+
+ // Partial window update (+3 bytes): sends "fgh", trailer still deferred
+ ctx->_remote_window_left.fetch_add(3, butil::memory_order_relaxed);
+ sctx->_remote_window_left.fetch_add(3, butil::memory_order_relaxed);
+ {
+ std::unique_lock<butil::Mutex> mu(ctx->_stream_mutex);
+ ctx->AppendPendingDataLocked(sctx, &out);
+ }
+ frame = PopFrame(&out, &payload);
+ EXPECT_EQ(brpc::policy::H2_FRAME_DATA, frame.type);
+ EXPECT_EQ(3u, frame.payload_size);
+ EXPECT_EQ(0, frame.flags & H2_FLAGS_END_STREAM);
+ EXPECT_EQ("fgh", payload);
+ EXPECT_TRUE(out.empty());
+ EXPECT_EQ(2u, ctx->_pending_data_size.load());
+ EXPECT_FALSE(sctx->_pending_trailer.empty());
+
+ // Final window update (+2 bytes): sends "ij" followed by Trailer
HEADERS(END_STREAM)
+ ctx->_remote_window_left.fetch_add(2, butil::memory_order_relaxed);
+ sctx->_remote_window_left.fetch_add(2, butil::memory_order_relaxed);
+ {
+ std::unique_lock<butil::Mutex> mu(ctx->_stream_mutex);
+ ctx->AppendPendingDataLocked(sctx, &out);
+ }
+ frame = PopFrame(&out, &payload);
+ EXPECT_EQ(brpc::policy::H2_FRAME_DATA, frame.type);
+ EXPECT_EQ(2u, frame.payload_size);
+ EXPECT_EQ(0, frame.flags & H2_FLAGS_END_STREAM);
+ EXPECT_EQ("ij", payload);
+
+ frame = PopFrame(&out, &payload);
+ EXPECT_EQ(brpc::policy::H2_FRAME_HEADERS, frame.type);
+ EXPECT_NE(0, frame.flags & H2_FLAGS_END_STREAM);
+ EXPECT_EQ(1, frame.stream_id);
+ while ((frame.flags & H2_FLAGS_END_HEADERS) == 0) {
+ frame = PopFrame(&out, &payload);
+ EXPECT_EQ(brpc::policy::H2_FRAME_CONTINUATION, frame.type);
+ EXPECT_EQ(1, frame.stream_id);
+ }
+ EXPECT_TRUE(out.empty());
+ EXPECT_EQ(0u, ctx->_pending_data_size.load());
+}
+
+TEST(H2UnsentMessage, deferred_trailer_does_not_corrupt_cross_stream_hpack) {
+ brpc::Server server;
+ brpc::SocketId id;
+ brpc::SocketUniquePtr sock;
+ brpc::SocketOptions options;
+ ASSERT_EQ(0, brpc::Socket::Create(options, &id));
+ ASSERT_EQ(0, brpc::Socket::Address(id, &sock));
+
+ brpc::policy::H2Context* server_ctx =
+ new brpc::policy::H2Context(sock.get(), &server);
+ ASSERT_EQ(0, server_ctx->Init());
+ sock->initialize_parsing_context(&server_ctx);
+ server_ctx->_remote_settings.stream_window_size = 0;
+ server_ctx->_remote_window_left = 100;
+
+ // Stream 1: window is 0, so its DATA and gRPC Trailer are deferred
+ brpc::policy::H2StreamContext* sctx1 =
+ new brpc::policy::H2StreamContext(false);
+ sctx1->Init(server_ctx, 1);
+ ASSERT_EQ(0, server_ctx->TryToInsertStream(1, sctx1));
+ sctx1->_remote_ended = true;
+
+ // Stream 3: has sufficient window (20 bytes), sends HEADERS + DATA +
Trailer immediately
+ brpc::policy::H2StreamContext* sctx3 =
+ new brpc::policy::H2StreamContext(false);
+ sctx3->Init(server_ctx, 3);
+ ASSERT_EQ(0, server_ctx->TryToInsertStream(3, sctx3));
+ sctx3->_remote_window_left = 20;
+ sctx3->_remote_ended = true;
+
+ butil::IOBuf wire_buf;
+ {
+ brpc::Controller cntl1;
+ cntl1.http_response().set_content_type("application/grpc");
+ cntl1.http_response().SetHeader("x-custom-stream", "stream-one");
+ cntl1.response_attachment().append("body1");
+ brpc::policy::H2UnsentResponse* res1 =
+ brpc::policy::H2UnsentResponse::New(&cntl1, 1, true);
+ ASSERT_TRUE(res1->AppendAndDestroySelf(&wire_buf, sock.get()).ok());
+ }
+ {
+ brpc::Controller cntl3;
+ cntl3.http_response().set_content_type("application/grpc");
+ cntl3.http_response().SetHeader("x-custom-stream", "stream-three");
+ cntl3.response_attachment().append("body3");
+ brpc::policy::H2UnsentResponse* res3 =
+ brpc::policy::H2UnsentResponse::New(&cntl3, 3, true);
+ ASSERT_TRUE(res3->AppendAndDestroySelf(&wire_buf, sock.get()).ok());
+ }
+ EXPECT_EQ(nullptr, server_ctx->FindStream(3));
+ ASSERT_NE(nullptr, server_ctx->FindStream(1));
+
+ // Now replenish Stream 1's window and flush its DATA + deferred Trailer
+ sctx1->_remote_window_left.fetch_add(10, butil::memory_order_relaxed);
+ {
+ std::unique_lock<butil::Mutex> mu(server_ctx->_stream_mutex);
+ server_ctx->AppendPendingDataLocked(sctx1, &wire_buf);
+ }
+
+ // Decode all HEADERS frames in wire transmission order using a client
HPacker
+ brpc::HPacker client_hpacker;
+ ASSERT_EQ(0,
client_hpacker.Init(brpc::H2Settings::DEFAULT_HEADER_TABLE_SIZE));
+ int trailer_stream1_count = 0;
+ while (!wire_buf.empty()) {
+ std::string payload;
+ brpc::policy::H2FrameHead frame = PopFrame(&wire_buf, &payload);
+ if (frame.type == brpc::policy::H2_FRAME_HEADERS) {
+ butil::IOBuf hdr_buf;
+ hdr_buf.append(payload);
+ butil::IOBufBytesIterator it(hdr_buf);
+ while (it != NULL) {
+ brpc::HPacker::Header h;
+ ASSERT_GT(client_hpacker.Decode(it, &h), 0);
+ if (frame.stream_id == 1 &&
+ (frame.flags & H2_FLAGS_END_STREAM) &&
+ h.name == "grpc-status") {
+ EXPECT_EQ("0", h.value);
+ ++trailer_stream1_count;
+ }
+ }
+ }
+ }
+ EXPECT_EQ(1, trailer_stream1_count);
+}
+
+TEST(H2UnsentMessage, server_pending_response_cleaned_up_on_rst_stream) {
+ brpc::Server server;
+ brpc::SocketId id;
+ brpc::SocketUniquePtr sock;
+ brpc::SocketOptions options;
+ ASSERT_EQ(0, brpc::Socket::Create(options, &id));
+ ASSERT_EQ(0, brpc::Socket::Address(id, &sock));
+
+ brpc::policy::H2Context* ctx =
+ new brpc::policy::H2Context(sock.get(), &server);
+ ASSERT_EQ(0, ctx->Init());
+ sock->initialize_parsing_context(&ctx);
+ ctx->_remote_window_left = 0;
+
+ brpc::policy::H2StreamContext* sctx =
+ new brpc::policy::H2StreamContext(false);
+ sctx->Init(ctx, 1);
+ ASSERT_EQ(0, ctx->TryToInsertStream(1, sctx));
+ sctx->_remote_ended = true;
+
+ brpc::Controller cntl;
+ cntl.http_response().set_content_type("application/grpc");
+ cntl.response_attachment().append("0123456789");
+ brpc::policy::H2UnsentResponse* response =
+ brpc::policy::H2UnsentResponse::New(&cntl, 1, true);
+ butil::IOBuf out;
+ ASSERT_TRUE(response->AppendAndDestroySelf(&out, sock.get()).ok());
+ EXPECT_EQ(10u, ctx->_pending_data_size.load());
+ ASSERT_NE(nullptr, ctx->FindStream(1));
+
+ // Simulate client sending RST_STREAM(CANCEL) for stream 1
+ const char rst_payload[] = {0, 0, 0, 8 /* H2_CANCEL */};
+ butil::IOBuf payload_buf;
+ payload_buf.append(rst_payload, sizeof(rst_payload));
+ butil::IOBufBytesIterator it(payload_buf);
+ const brpc::policy::H2FrameHead rst_head = {
+ 4, brpc::policy::H2_FRAME_RST_STREAM, 0, 1};
+ brpc::policy::H2ParseResult res = ctx->OnResetStream(it, rst_head);
+ EXPECT_TRUE(res.is_ok());
+ EXPECT_EQ(nullptr, ctx->FindStream(1));
+ EXPECT_EQ(0u, ctx->_pending_data_size.load());
+}
+
+TEST(H2UnsentMessage, server_pending_response_overcrowded_sends_rst_stream) {
+ brpc::Server server;
+ brpc::SocketId id;
+ brpc::SocketUniquePtr sock;
+ brpc::SocketOptions options;
+ ASSERT_EQ(0, brpc::Socket::Create(options, &id));
+ ASSERT_EQ(0, brpc::Socket::Address(id, &sock));
+
+ brpc::policy::H2Context* ctx =
+ new brpc::policy::H2Context(sock.get(), &server);
+ ASSERT_EQ(0, ctx->Init());
+ sock->initialize_parsing_context(&ctx);
+ ctx->_remote_window_left = 0;
+
+ GFLAGS_NAMESPACE::FlagSaver flag_saver;
+ brpc::FLAGS_socket_max_unwritten_bytes = 8;
+
+ // Stream 1 queues 8 bytes, reaching socket_max_unwritten_bytes
+ brpc::policy::H2StreamContext* sctx1 =
+ new brpc::policy::H2StreamContext(false);
+ sctx1->Init(ctx, 1);
+ ASSERT_EQ(0, ctx->TryToInsertStream(1, sctx1));
+ sctx1->_remote_ended = true;
+
+ brpc::Controller cntl1;
+ cntl1.response_attachment().append("12345678");
+ brpc::policy::H2UnsentResponse* res1 =
+ brpc::policy::H2UnsentResponse::New(&cntl1, 1, false);
+ butil::IOBuf out1;
+ ASSERT_TRUE(res1->AppendAndDestroySelf(&out1, sock.get()).ok());
+ EXPECT_EQ(8u, ctx->_pending_data_size.load());
+ EXPECT_TRUE(ctx->PendingDataOvercrowded());
+
+ // Stream 3 arrives while overcrowded -> emits
RST_STREAM(ENHANCE_YOUR_CALM)
+ brpc::policy::H2StreamContext* sctx3 =
+ new brpc::policy::H2StreamContext(false);
+ sctx3->Init(ctx, 3);
+ ASSERT_EQ(0, ctx->TryToInsertStream(3, sctx3));
+ sctx3->_remote_ended = true;
+
+ brpc::Controller cntl3;
+ cntl3.response_attachment().append("abcdefgh");
+ brpc::policy::H2UnsentResponse* res3 =
+ brpc::policy::H2UnsentResponse::New(&cntl3, 3, true);
+ butil::IOBuf out3;
+ ASSERT_TRUE(res3->AppendAndDestroySelf(&out3, sock.get()).ok());
+
+ std::string payload;
+ brpc::policy::H2FrameHead frame = PopFrame(&out3, &payload);
+ EXPECT_EQ(brpc::policy::H2_FRAME_RST_STREAM, frame.type);
+ EXPECT_EQ(3, frame.stream_id);
+ EXPECT_EQ(4u, frame.payload_size);
+ EXPECT_TRUE(out3.empty());
+ EXPECT_EQ(nullptr, ctx->FindStream(3));
+ EXPECT_EQ(8u, ctx->_pending_data_size.load());
+}
+
+TEST(H2UnsentMessage, flush_pending_data_round_robin_fairness) {
+ brpc::Server server;
+ int fds[2];
+ ASSERT_EQ(0, pipe(fds));
+ brpc::SocketId id;
+ brpc::SocketUniquePtr sock;
+ brpc::SocketOptions options;
+ options.fd = fds[1];
+ ASSERT_EQ(0, brpc::Socket::Create(options, &id));
+ ASSERT_EQ(0, brpc::Socket::Address(id, &sock));
+
+ brpc::policy::H2Context* ctx =
+ new brpc::policy::H2Context(sock.get(), &server);
+ ASSERT_EQ(0, ctx->Init());
+ sock->initialize_parsing_context(&ctx);
+ ctx->_remote_settings.max_frame_size = 4;
+ ctx->_remote_settings.stream_window_size = 16;
+ ctx->_remote_window_left = 0;
+
+ for (int stream_id : {1, 3}) {
+ brpc::policy::H2StreamContext* sctx =
+ new brpc::policy::H2StreamContext(false);
+ sctx->Init(ctx, stream_id);
+ ASSERT_EQ(0, ctx->TryToInsertStream(stream_id, sctx));
+ sctx->_remote_ended = true;
+
+ brpc::Controller cntl;
+ cntl.response_attachment().append(stream_id == 1 ? "11112222" :
"33334444");
+ brpc::policy::H2UnsentResponse* res =
+ brpc::policy::H2UnsentResponse::New(&cntl, stream_id, false);
+ butil::IOBuf hdr_out;
+ ASSERT_TRUE(res->AppendAndDestroySelf(&hdr_out, sock.get()).ok());
+ }
+ EXPECT_EQ(16u, ctx->_pending_data_size.load());
+
+ // Replenish connection window by 4 bytes -> FlushPendingData flushes
stream 1 ("1111")
+ ctx->_remote_window_left.fetch_add(4, butil::memory_order_relaxed);
+ ASSERT_TRUE(ctx->FlushPendingData(0));
+ EXPECT_EQ(1, ctx->_last_flushed_stream_id);
+ EXPECT_EQ(12u, ctx->_pending_data_size.load());
+
+ // Replenish connection window by 4 bytes -> Round-robin flushes stream 3
("3333") next!
+ ctx->_remote_window_left.fetch_add(4, butil::memory_order_relaxed);
+ ASSERT_TRUE(ctx->FlushPendingData(0));
+ EXPECT_EQ(3, ctx->_last_flushed_stream_id);
+ EXPECT_EQ(8u, ctx->_pending_data_size.load());
+
+ // Replenish connection window by 8 bytes -> Flushes remaining stream 1
("2222") and stream 3 ("4444")
+ ctx->_remote_window_left.fetch_add(8, butil::memory_order_relaxed);
+ ASSERT_TRUE(ctx->FlushPendingData(0));
+ EXPECT_EQ(0u, ctx->_pending_data_size.load());
+ EXPECT_EQ(nullptr, ctx->FindStream(1));
+ EXPECT_EQ(nullptr, ctx->FindStream(3));
+
+ butil::IOPortal pipe_buf;
+ ASSERT_GT(pipe_buf.append_from_file_descriptor(fds[0], 1024), 0);
+ std::string payload;
+ brpc::policy::H2FrameHead f1 = PopFrame(&pipe_buf, &payload);
+ EXPECT_EQ(1, f1.stream_id);
+ EXPECT_EQ("1111", payload);
+ brpc::policy::H2FrameHead f2 = PopFrame(&pipe_buf, &payload);
+ EXPECT_EQ(3, f2.stream_id);
+ EXPECT_EQ("3333", payload);
+ brpc::policy::H2FrameHead f3 = PopFrame(&pipe_buf, &payload);
+ EXPECT_EQ(1, f3.stream_id);
+ EXPECT_EQ("2222", payload);
+ EXPECT_NE(0, f3.flags & H2_FLAGS_END_STREAM);
+ brpc::policy::H2FrameHead f4 = PopFrame(&pipe_buf, &payload);
+ EXPECT_EQ(3, f4.stream_id);
+ EXPECT_EQ("4444", payload);
+ EXPECT_NE(0, f4.flags & H2_FLAGS_END_STREAM);
+ close(fds[0]);
+}
+
+TEST(H2UnsentMessage,
client_rejected_request_does_not_corrupt_preface_or_hpack) {
+ GFLAGS_NAMESPACE::FlagSaver flag_saver;
+ brpc::FLAGS_socket_max_unwritten_bytes = 4;
+
+ brpc::SocketId client_id;
+ brpc::SocketUniquePtr client_sock;
+ brpc::SocketOptions client_options;
+ client_options.user = brpc::get_client_side_messenger();
+ // CreatedByConnect() is true when remote_side is non-empty.
+ ASSERT_EQ(0, butil::str2endpoint("127.0.0.1:8010",
&client_options.remote_side));
+ ASSERT_EQ(0, brpc::Socket::Create(client_options, &client_id));
+ ASSERT_EQ(0, brpc::Socket::Address(client_id, &client_sock));
+
+ // First request on brand-new connection exceeds (initial_window=65535 +
max_unwritten=4).
+ brpc::Controller cntl1;
+ cntl1.http_request().uri() = "http://example.com/echo";;
+ cntl1.http_request().SetHeader("x-custom-dynamic-header",
"first-rejected-value");
+ cntl1.request_attachment().append(std::string(65535 + 16, 'x'));
+ brpc::policy::H2UnsentRequest* req1 =
+ brpc::policy::H2UnsentRequest::New(&cntl1);
+ butil::IOBuf out1;
+ butil::Status st1 = req1->AppendAndDestroySelf(&out1, client_sock.get());
+ EXPECT_EQ(brpc::EOVERCROWDED, st1.error_code());
+ EXPECT_TRUE(out1.empty());
+
+ brpc::policy::H2Context* client_ctx =
+ static_cast<brpc::policy::H2Context*>(client_sock->parsing_context());
+ ASSERT_NE(nullptr, client_ctx);
+#if defined(UNIT_TEST)
+ const int initial_stream_id = 0x7fffffff - 10000;
+#else
+ const int initial_stream_id = 1;
+#endif
+ // Stream ID must not have been consumed, and preface must still be
pending.
+ EXPECT_EQ(initial_stream_id, client_ctx->_last_sent_stream_id);
+ EXPECT_TRUE(client_ctx->_client_preface_pending);
+
+ // Second request fits within window and must emit connection preface +
SETTINGS + valid HPACK.
+ brpc::Controller cntl2;
+ cntl2.http_request().uri() = "http://example.com/echo";;
+ cntl2.http_request().SetHeader("x-custom-dynamic-header",
"first-rejected-value");
+ cntl2.request_attachment().append("ok");
+ brpc::policy::H2UnsentRequest* req2 =
+ brpc::policy::H2UnsentRequest::New(&cntl2);
+ butil::IOBuf out2;
+ butil::Status st2 = req2->AppendAndDestroySelf(&out2, client_sock.get());
+ ASSERT_TRUE(st2.ok()) << st2;
+ EXPECT_EQ(initial_stream_id + 2, client_ctx->_last_sent_stream_id);
+ EXPECT_FALSE(client_ctx->_client_preface_pending);
+
+ // Verify a fresh server H2Context can parse out2 (including preface,
SETTINGS, and HPACK headers).
+ brpc::Server server;
+ int fds[2];
+ ASSERT_EQ(0, pipe(fds));
+ brpc::SocketId server_id;
+ brpc::SocketUniquePtr server_sock;
+ brpc::SocketOptions server_options;
+ server_options.fd = fds[1];
+ ASSERT_EQ(0, brpc::Socket::Create(server_options, &server_id));
+ ASSERT_EQ(0, brpc::Socket::Address(server_id, &server_sock));
+
+ brpc::ParseResult pr = brpc::policy::ParseH2Message(
+ &out2, server_sock.get(), false, &server);
+ ASSERT_TRUE(pr.is_ok());
+ ASSERT_NE(nullptr, pr.message());
+ auto* received_sctx =
+ static_cast<brpc::policy::H2StreamContext*>(pr.message());
+ EXPECT_EQ(initial_stream_id, received_sctx->stream_id());
+ const std::string* hdr_val =
+ received_sctx->header().GetHeader("x-custom-dynamic-header");
+ ASSERT_NE(nullptr, hdr_val);
+ EXPECT_EQ("first-rejected-value", *hdr_val);
+ received_sctx->Destroy();
+ close(fds[0]);
+}
+
+TEST(H2UnsentMessage,
rejected_headers_on_remote_ended_stream_preserves_frame_and_hpack) {
+ brpc::Server server;
+ int fds[2];
+ ASSERT_EQ(0, pipe(fds));
+ brpc::SocketId server_id;
+ brpc::SocketUniquePtr server_sock;
+ brpc::SocketOptions server_options;
+ server_options.fd = fds[1];
+ ASSERT_EQ(0, brpc::Socket::Create(server_options, &server_id));
+ ASSERT_EQ(0, brpc::Socket::Address(server_id, &server_sock));
+
+ brpc::policy::H2Context* srv_ctx =
+ new brpc::policy::H2Context(server_sock.get(), &server);
+ ASSERT_EQ(0, srv_ctx->Init());
+ server_sock->initialize_parsing_context(&srv_ctx);
+ srv_ctx->_conn_state = brpc::policy::H2_CONNECTION_READY;
+
+ // Insert Stream 1 as already remote-ended (retained while response is
pending).
+ brpc::policy::H2StreamContext* sctx1 =
+ new brpc::policy::H2StreamContext(false);
+ sctx1->Init(srv_ctx, 1);
+ sctx1->_remote_ended = true;
+ ASSERT_EQ(0, srv_ctx->TryToInsertStream(1, sctx1));
+
+ brpc::HPacker client_hpacker;
+ ASSERT_EQ(0,
client_hpacker.Init(brpc::H2Settings::DEFAULT_HEADER_TABLE_SIZE));
+ brpc::HPackOptions hpack_opt;
+ hpack_opt.encode_name = false;
+ hpack_opt.encode_value = false;
+
+ // Encode a fragmented header block for Stream 1 across HEADERS (without
+ // END_HEADERS) + CONTINUATION (with END_HEADERS) that inserts entries into
+ // the HPACK dynamic table.
+ butil::IOBufAppender appender1_part1;
+ client_hpacker.Encode(
+ &appender1_part1, brpc::HPacker::Header("x-rejected-first",
"first-val"), hpack_opt);
+ butil::IOBuf frag1_part1;
+ appender1_part1.move_to(frag1_part1);
+
+ butil::IOBufAppender appender1_part2;
+ client_hpacker.Encode(
+ &appender1_part2, brpc::HPacker::Header("x-rejected-dyn",
"dyn-value-123"), hpack_opt);
+ butil::IOBuf frag1_part2;
+ appender1_part2.move_to(frag1_part2);
+
+ butil::IOBuf wire;
+ char headbuf[brpc::policy::FRAME_HEAD_SIZE];
+ brpc::policy::SerializeFrameHead(
+ headbuf, frag1_part1.size(), brpc::policy::H2_FRAME_HEADERS, 0, 1);
+ wire.append(headbuf, sizeof(headbuf));
+ wire.append(frag1_part1);
+
+ brpc::policy::SerializeFrameHead(
+ headbuf, frag1_part2.size(), brpc::policy::H2_FRAME_CONTINUATION,
+ H2_FLAGS_END_HEADERS, 1);
+ wire.append(headbuf, sizeof(headbuf));
+ wire.append(frag1_part2);
+
+ // Immediately follow with a new valid request on Stream 3 referencing the
+ // dynamic table entry added by the rejected Stream 1 header block.
+ butil::IOBufAppender appender3;
+ client_hpacker.Encode(&appender3, brpc::HPacker::Header(":method", "GET"),
hpack_opt);
+ client_hpacker.Encode(&appender3, brpc::HPacker::Header(":scheme",
"http"), hpack_opt);
+ client_hpacker.Encode(&appender3, brpc::HPacker::Header(":path",
"/EchoService/Echo"), hpack_opt);
+ client_hpacker.Encode(
+ &appender3, brpc::HPacker::Header("x-rejected-dyn", "dyn-value-123"),
hpack_opt);
+ butil::IOBuf hdr_block3;
+ appender3.move_to(hdr_block3);
+
+ brpc::policy::SerializeFrameHead(
+ headbuf, hdr_block3.size(), brpc::policy::H2_FRAME_HEADERS,
+ H2_FLAGS_END_HEADERS | H2_FLAGS_END_STREAM, 3);
+ wire.append(headbuf, sizeof(headbuf));
+ wire.append(hdr_block3);
+
+ // Parse the combined wire buffer: Stream 1 must be reset with
STREAM_CLOSED
+ // after CONTINUATION, and Stream 3 must parse cleanly with the updated
HPACK table.
+ brpc::ParseResult pr = brpc::policy::ParseH2Message(
+ &wire, server_sock.get(), false, &server);
+ ASSERT_TRUE(pr.is_ok());
+ ASSERT_NE(nullptr, pr.message());
+ EXPECT_EQ(nullptr, srv_ctx->FindStream(1));
+
+ auto* sctx3 = static_cast<brpc::policy::H2StreamContext*>(pr.message());
+ EXPECT_EQ(3, sctx3->stream_id());
+ const std::string* val = sctx3->header().GetHeader("x-rejected-dyn");
+ ASSERT_NE(nullptr, val);
+ EXPECT_EQ("dyn-value-123", *val);
+ sctx3->Destroy();
+ close(fds[0]);
+}
+
+TEST(H2UnsentMessage,
deferred_response_notify_registration_failure_resets_stream) {
+ brpc::Server server;
+ int fds[2];
+ ASSERT_EQ(0, pipe(fds));
+ brpc::SocketId server_id;
+ brpc::SocketUniquePtr server_sock;
+ brpc::SocketOptions server_options;
+ server_options.fd = fds[1];
+ ASSERT_EQ(0, brpc::Socket::Create(server_options, &server_id));
+ ASSERT_EQ(0, brpc::Socket::Address(server_id, &server_sock));
+
+ brpc::policy::H2Context* srv_ctx =
+ new brpc::policy::H2Context(server_sock.get(), &server);
+ ASSERT_EQ(0, srv_ctx->Init());
+ server_sock->initialize_parsing_context(&srv_ctx);
+ srv_ctx->_conn_state = brpc::policy::H2_CONNECTION_READY;
+ srv_ctx->_remote_settings.stream_window_size = 0;
+
+ brpc::policy::H2StreamContext* sctx =
+ new brpc::policy::H2StreamContext(false);
+ sctx->Init(srv_ctx, 1);
+ sctx->_remote_ended = true;
+ sctx->_deferred_window_update.store(8, butil::memory_order_relaxed);
+ ASSERT_EQ(0, srv_ctx->TryToInsertStream(1, sctx));
+
+ // Saturate server_sock->_id_wait_list so Socket::NotifyOnFailed() fails
with
+ // EAGAIN while server_sock itself remains healthy
(!server_sock->Failed()).
+ bthread_id_t dummy_id = INVALID_BTHREAD_ID;
+ ASSERT_EQ(0, bthread_id_create(&dummy_id, NULL, NULL));
+ {
+ BAIDU_SCOPED_LOCK(server_sock->_id_wait_list_mutex);
+ int consecutive_eagain = 0;
+ while (consecutive_eagain < 26000) {
+ const int rc =
+ bthread_id_list_add(&server_sock->_id_wait_list, dummy_id);
+ if (rc == EAGAIN) {
+ ++consecutive_eagain;
+ } else {
+ ASSERT_EQ(0, rc);
+ consecutive_eagain = 0;
+ }
+ }
+ }
+ ASSERT_FALSE(server_sock->Failed());
+
+ brpc::Controller cntl;
+ cntl.http_response().set_status_code(brpc::HTTP_STATUS_OK);
+ cntl.response_attachment().append("hello world");
+ brpc::policy::H2UnsentResponse* resp =
+ brpc::policy::H2UnsentResponse::New(&cntl, 1, false);
+ ASSERT_NE(nullptr, resp);
+
+ butil::IOBuf out;
+ butil::Status st = resp->AppendAndDestroySelf(&out, server_sock.get());
+ ASSERT_TRUE(st.ok()) << st;
+
+ EXPECT_FALSE(server_sock->Failed());
+ EXPECT_EQ(nullptr, srv_ctx->FindStream(1));
+ EXPECT_EQ(0u, srv_ctx->_pending_data_size.load());
+ EXPECT_EQ(0, server_sock->_idle_fail_ref_count.load());
+
+ std::string payload;
+ brpc::policy::H2FrameHead frame = PopFrame(&out, &payload);
+ EXPECT_EQ(brpc::policy::H2_FRAME_HEADERS, frame.type);
+ EXPECT_EQ(1, frame.stream_id);
+ EXPECT_EQ(0, frame.flags & H2_FLAGS_END_STREAM);
+
+ frame = PopFrame(&out, &payload);
+ EXPECT_EQ(brpc::policy::H2_FRAME_RST_STREAM, frame.type);
+ EXPECT_EQ(1, frame.stream_id);
+ ASSERT_EQ(4u, payload.size());
+ const uint32_t rst_err =
+ (static_cast<uint8_t>(payload[0]) << 24) |
+ (static_cast<uint8_t>(payload[1]) << 16) |
+ (static_cast<uint8_t>(payload[2]) << 8) |
+ static_cast<uint8_t>(payload[3]);
+ EXPECT_EQ(static_cast<uint32_t>(brpc::H2_INTERNAL_ERROR), rst_err);
+
+ frame = PopFrame(&out, &payload);
+ EXPECT_EQ(brpc::policy::H2_FRAME_WINDOW_UPDATE, frame.type);
+ EXPECT_EQ(0, frame.stream_id);
+ EXPECT_TRUE(out.empty());
+
+ EXPECT_EQ(0, bthread_id_error(dummy_id, 0));
+ close(fds[0]);
+}
+
+TEST(H2UnsentMessage, early_response_migrates_deferred_window_update) {
+ brpc::Server server;
+ int fds[2];
+ ASSERT_EQ(0, pipe(fds));
+ brpc::SocketId server_id;
+ brpc::SocketUniquePtr server_sock;
+ brpc::SocketOptions server_options;
+ server_options.fd = fds[1];
+ ASSERT_EQ(0, brpc::Socket::Create(server_options, &server_id));
+ ASSERT_EQ(0, brpc::Socket::Address(server_id, &server_sock));
+
+ brpc::policy::H2Context* srv_ctx =
+ new brpc::policy::H2Context(server_sock.get(), &server);
+ ASSERT_EQ(0, srv_ctx->Init());
+ server_sock->initialize_parsing_context(&srv_ctx);
+ srv_ctx->_conn_state = brpc::policy::H2_CONNECTION_READY;
+
+ // Case 1: Immediate response completion in
H2UnsentResponse::AppendAndDestroySelf
+ // when the request has not ended yet (_remote_ended == false) must migrate
+ // sctx->_deferred_window_update to the connection and emit
connection-level
+ // WINDOW_UPDATE in `out'.
+ brpc::policy::H2StreamContext* sctx1 =
+ new brpc::policy::H2StreamContext(true);
+ sctx1->Init(srv_ctx, 1);
+ sctx1->_remote_ended = false;
+ sctx1->_deferred_window_update.store(1234, butil::memory_order_relaxed);
+ ASSERT_EQ(0, srv_ctx->TryToInsertStream(1, sctx1));
+
+ {
+ brpc::Controller cntl;
+ cntl.http_response().set_status_code(brpc::HTTP_STATUS_OK);
+ cntl.response_attachment().append("ok");
+ brpc::policy::H2UnsentResponse* resp =
+ brpc::policy::H2UnsentResponse::New(&cntl, 1, false);
+ ASSERT_NE(nullptr, resp);
+ butil::IOBuf out;
+ ASSERT_TRUE(resp->AppendAndDestroySelf(&out, server_sock.get()).ok());
+ EXPECT_EQ(nullptr, srv_ctx->FindStream(1));
+
+ std::string payload;
+ brpc::policy::H2FrameHead frame = PopFrame(&out, &payload);
+ EXPECT_EQ(brpc::policy::H2_FRAME_HEADERS, frame.type);
+ frame = PopFrame(&out, &payload);
+ EXPECT_EQ(brpc::policy::H2_FRAME_DATA, frame.type);
+ EXPECT_NE(0, frame.flags & H2_FLAGS_END_STREAM);
+ frame = PopFrame(&out, &payload);
+ EXPECT_EQ(brpc::policy::H2_FRAME_WINDOW_UPDATE, frame.type);
+ EXPECT_EQ(0, frame.stream_id);
+ ASSERT_EQ(4u, payload.size());
+ const uint32_t wu_inc =
+ (static_cast<uint8_t>(payload[0]) << 24) |
+ (static_cast<uint8_t>(payload[1]) << 16) |
+ (static_cast<uint8_t>(payload[2]) << 8) |
+ static_cast<uint8_t>(payload[3]);
+ EXPECT_EQ(1234u, wu_inc);
+ EXPECT_TRUE(out.empty());
+ }
+
+ // Case 2: Deferred response completion in H2Context::FlushPendingData when
+ // the request has not ended yet (_remote_ended == false) must also migrate
+ // sctx->_deferred_window_update to the connection.
+ srv_ctx->_remote_settings.stream_window_size = 0;
+ brpc::policy::H2StreamContext* sctx3 =
+ new brpc::policy::H2StreamContext(true);
+ sctx3->Init(srv_ctx, 3);
+ sctx3->_remote_ended = false;
+ sctx3->_deferred_window_update.store(4321, butil::memory_order_relaxed);
+ ASSERT_EQ(0, srv_ctx->TryToInsertStream(3, sctx3));
+
+ {
+ brpc::Controller cntl;
+ cntl.http_response().set_status_code(brpc::HTTP_STATUS_OK);
+ cntl.response_attachment().append("deferred");
+ brpc::policy::H2UnsentResponse* resp =
+ brpc::policy::H2UnsentResponse::New(&cntl, 3, false);
+ ASSERT_NE(nullptr, resp);
+ butil::IOBuf out;
+ ASSERT_TRUE(resp->AppendAndDestroySelf(&out, server_sock.get()).ok());
+ ASSERT_NE(nullptr, srv_ctx->FindStream(3));
+ // Clear the connection WU flushed during AppendAndDestroySelf, then
+ // restore sctx3's request-side deferred window credit to verify
+ // FlushPendingData's own migration path.
+ EXPECT_EQ(0, srv_ctx->_deferred_window_update.exchange(
+ 0, butil::memory_order_relaxed));
+ sctx3->_deferred_window_update.store(4321,
butil::memory_order_relaxed);
+
+ sctx3->_remote_window_left.store(64, butil::memory_order_relaxed);
+ ASSERT_TRUE(srv_ctx->FlushPendingData(3));
+ EXPECT_EQ(nullptr, srv_ctx->FindStream(3));
+ EXPECT_EQ(4321, srv_ctx->_deferred_window_update.exchange(
+ 0, butil::memory_order_relaxed));
+ }
+ close(fds[0]);
+}
diff --git a/test/brpc_http_rpc_protocol_unittest.cpp
b/test/brpc_http_rpc_protocol_unittest.cpp
index f3e07f4a8..333950568 100644
--- a/test/brpc_http_rpc_protocol_unittest.cpp
+++ b/test/brpc_http_rpc_protocol_unittest.cpp
@@ -23,6 +23,7 @@
#include <cstddef>
#include <set>
#include <string>
+#include <thread>
#include <sys/ioctl.h>
#include <sys/socket.h>
#include <butil/build_config.h> // OS_MACOSX
@@ -69,6 +70,9 @@ DECLARE_int64(socket_max_unwritten_bytes);
DECLARE_uint32(http_max_header_count);
DECLARE_uint32(http_max_query_count);
extern bvar::CollectorSpeedLimit g_rpc_dump_sl;
+namespace policy {
+DECLARE_int64(h2_max_pending_data_bytes);
+}
}
int main(int argc, char* argv[]) {
@@ -330,6 +334,21 @@ protected:
}
void MakeH2EchoResponseBuf(butil::IOBuf* out, int h2_stream_id) {
+ brpc::policy::H2Context* srv_ctx =
+ static_cast<brpc::policy::H2Context*>(_socket->parsing_context());
+ if (srv_ctx == NULL) {
+ srv_ctx = new brpc::policy::H2Context(_socket.get(), &_server);
+ ASSERT_EQ(0, srv_ctx->Init());
+ _socket->initialize_parsing_context(&srv_ctx);
+ }
+ if (srv_ctx->FindStream(h2_stream_id) == NULL) {
+ brpc::policy::H2StreamContext* srv_sctx =
+ new brpc::policy::H2StreamContext(false);
+ srv_sctx->Init(srv_ctx, h2_stream_id);
+ srv_sctx->_remote_ended = true;
+ ASSERT_EQ(0, srv_ctx->TryToInsertStream(h2_stream_id, srv_sctx));
+ }
+
brpc::Controller cntl;
test::EchoResponse res;
res.set_message(EXP_RESPONSE);
@@ -339,7 +358,7 @@ protected:
EXPECT_TRUE(res.SerializeToZeroCopyStream(&wrapper));
}
brpc::policy::H2UnsentResponse* h2_res =
brpc::policy::H2UnsentResponse::New(&cntl, h2_stream_id, false /*is grpc*/);
- butil::Status st = h2_res->AppendAndDestroySelf(out,
_h2_client_sock.get());
+ butil::Status st = h2_res->AppendAndDestroySelf(out, _socket.get());
ASSERT_TRUE(st.ok());
}
@@ -1714,10 +1733,15 @@ TEST_F(HttpTest, http2_sanity) {
// Check that the first request larger than the default window completes
// after SETTINGS and WINDOW_UPDATE make more capacity available.
+ // main() sets -socket_max_unwritten_bytes=2000000 for HTTP/1 overcrowding
+ // tests, so raise -h2_max_pending_data_bytes to allow buffering the 2MB
+ // HTTP/2 request before the initial SETTINGS/WINDOW_UPDATE arrives.
+ GFLAGS_NAMESPACE::FlagSaver flag_saver;
+ brpc::policy::FLAGS_h2_max_pending_data_bytes = 4 * 1024 * 1024;
brpc::Controller cntl;
test::EchoRequest big_req;
test::EchoResponse res;
- std::string message(128 * 1024, 'x');
+ std::string message(2 * 1024 * 1024 /* 2M */, 'x');
big_req.set_message(message);
cntl.http_request().set_method(brpc::HTTP_METHOD_POST);
cntl.http_request().uri() = "/EchoService/Echo";
@@ -1806,6 +1830,33 @@ TEST_F(HttpTest, http2_rst_before_header) {
ASSERT_TRUE(cntl.Failed());
ASSERT_TRUE(cntl.ErrorCode() == brpc::EHTTP);
ASSERT_TRUE(cntl.http_response().status_code() ==
brpc::HTTP_STATUS_INTERNAL_SERVER_ERROR);
+ ASSERT_NE(std::string::npos,
+ cntl.ErrorText().find("terminated by RST_STREAM:
INTERNAL_ERROR[0x2]"))
+ << cntl.ErrorText();
+}
+
+TEST_F(HttpTest, http2_rst_flow_control_error_message) {
+ brpc::Controller cntl;
+ butil::IOBuf req_out;
+ int h2_stream_id = 0;
+ MakeH2EchoRequestBuf(&req_out, &cntl, &h2_stream_id);
+ butil::IOBuf res_out;
+ char rstbuf[9 /*FRAME_HEAD_SIZE*/ + 4];
+ brpc::policy::SerializeFrameHead(
+ rstbuf, 4, brpc::policy::H2_FRAME_RST_STREAM, 0, h2_stream_id);
+ SaveUint32(rstbuf + 9, brpc::H2_FLOW_CONTROL_ERROR);
+ res_out.append(rstbuf, sizeof(rstbuf));
+ brpc::ParseResult res_pr =
+ brpc::policy::ParseH2Message(&res_out, _h2_client_sock.get(), false,
NULL);
+ ASSERT_TRUE(res_pr.is_ok());
+ ProcessMessage(brpc::policy::ProcessHttpResponse, res_pr.message(), false);
+ ASSERT_TRUE(cntl.Failed());
+ EXPECT_EQ(brpc::EHTTP, cntl.ErrorCode());
+ EXPECT_EQ(brpc::HTTP_STATUS_INTERNAL_SERVER_ERROR,
+ cntl.http_response().status_code());
+ EXPECT_NE(std::string::npos,
+ cntl.ErrorText().find("terminated by RST_STREAM:
FLOW_CONTROL_ERROR[0x3]"))
+ << cntl.ErrorText();
}
TEST_F(HttpTest, http2_rst_after_header_and_data) {
@@ -1867,16 +1918,373 @@ TEST_F(HttpTest, http2_window_used_up_buffers_request)
{
butil::Status st = socket_message->AppendAndDestroySelf(&dummy,
_h2_client_sock.get());
ASSERT_TRUE(st.ok());
if (i == nsuc) {
- ASSERT_GT(ctx->_pending_data_size, 0u);
+ const int blocked_stream_id = h2_req->_stream_id;
+ ASSERT_GT(ctx->_pending_data_size.load(), 0u);
+ ASSERT_NE(nullptr, ctx->FindStream(blocked_stream_id));
h2_req->DestroyStreamUserData(
_h2_client_sock, &cntl, ECANCELED, false);
- ASSERT_EQ(0u, ctx->_pending_data_size);
+ ASSERT_EQ(0u, ctx->_pending_data_size.load());
+ EXPECT_EQ(nullptr, ctx->FindStream(blocked_stream_id));
} else {
h2_req->DestroyStreamUserData(_h2_client_sock, &cntl, 0, false);
}
}
}
+TEST_F(HttpTest,
http2_client_upload_timeout_sends_rst_and_cleans_server_stream) {
+ brpc::Server server;
+ EXPECT_EQ(0, server.AddService(&_svc, brpc::SERVER_DOESNT_OWN_SERVICE));
+ brpc::ServerOptions srv_options;
+ // Configure stream_window_size (256KB) > connection_window_size (64KB) so
a
+ // 100KB request sends HEADERS + the first ~64KB of DATA frames to the
server
+ // (stored in server sctx->body(), which is < quota=128KB so server does
not
+ // immediately send WINDOW_UPDATE), while the remaining ~36KB of DATA stays
+ // buffered in client _pending_data until the RPC times out mid-upload.
+ srv_options.h2_settings.stream_window_size = 256 * 1024;
+ srv_options.h2_settings.connection_window_size =
+ brpc::H2Settings::DEFAULT_INITIAL_WINDOW_SIZE;
+ ASSERT_EQ(0, server.Start(0, &srv_options));
+
+ brpc::Channel channel;
+ brpc::ChannelOptions options;
+ options.protocol = "h2";
+ ASSERT_EQ(0, channel.Init(server.listen_address(), &options));
+
+ // Warmup RPC ensures the connection is established and the client has
+ // received the server's SETTINGS frame.
+ {
+ brpc::Controller cntl;
+ test::EchoRequest req;
+ test::EchoResponse res;
+ req.set_message(EXP_REQUEST);
+ cntl.http_request().set_method(brpc::HTTP_METHOD_POST);
+ cntl.http_request().uri() = "/EchoService/Echo";
+ channel.CallMethod(NULL, &cntl, &req, &res, NULL);
+ ASSERT_FALSE(cntl.Failed()) << cntl.ErrorText();
+ ASSERT_EQ(EXP_RESPONSE, res.message());
+ }
+
+ auto read_bvar = [](const char* name) -> int64_t {
+ std::string val = bvar::Variable::describe_exposed(name);
+ int64_t out = -1;
+ butil::StringToInt64(val, &out);
+ return out;
+ };
+ const int64_t base_stream_count = read_bvar("h2_stream_context_count");
+ const int64_t base_rst_count = read_bvar("h2_stream_reset_count");
+
+ // Case 1: Send a 100KB POST request: ~64KB of DATA is uploaded to the
server
+ // and ~36KB remains buffered on the client waiting for connection
WINDOW_UPDATE.
+ // When it times out mid-upload, DestroyStreamUserData must send
+ // RST_STREAM(CANCEL) so the server releases its half-open H2StreamContext
+ // and partial request body.
+ {
+ brpc::Controller cntl;
+ test::EchoRequest req;
+ test::EchoResponse res;
+ req.set_message(std::string(100 * 1024, 'a'));
+ cntl.set_timeout_ms(50);
+ cntl.http_request().set_method(brpc::HTTP_METHOD_POST);
+ cntl.http_request().uri() = "/EchoService/Echo";
+ channel.CallMethod(NULL, &cntl, &req, &res, NULL);
+ ASSERT_TRUE(cntl.Failed());
+ EXPECT_EQ(brpc::ERPCTIMEDOUT, cntl.ErrorCode());
+ }
+
+ for (int i = 0; i < 40; ++i) {
+ if (read_bvar("h2_stream_context_count") == base_stream_count) {
+ break;
+ }
+ bthread_usleep(5000);
+ }
+ EXPECT_EQ(base_stream_count, read_bvar("h2_stream_context_count"));
+ EXPECT_EQ(0, read_bvar("h2_pending_data_bytes"));
+ // Both client (sending RST_STREAM) and server (receiving RST_STREAM in
+ // OnResetStream) increment h2_stream_reset_count in the same process.
+ EXPECT_EQ(base_rst_count + 2, read_bvar("h2_stream_reset_count"));
+
+ // Case 2: Small request completes upload (END_STREAM sent, client
_pending_data
+ // is empty), then times out while waiting for the server response.
+ // DestroyStreamUserData must still send RST_STREAM(CANCEL).
+ {
+ brpc::Controller cntl;
+ test::EchoRequest req;
+ test::EchoResponse res;
+ req.set_message(EXP_REQUEST);
+ cntl.set_timeout_ms(50);
+ cntl.http_request().set_method(brpc::HTTP_METHOD_POST);
+ cntl.http_request().uri() = "/EchoService/Echo?sleep_ms=200";
+ channel.CallMethod(NULL, &cntl, &req, &res, NULL);
+ ASSERT_TRUE(cntl.Failed());
+ EXPECT_EQ(brpc::ERPCTIMEDOUT, cntl.ErrorCode());
+ }
+ for (int i = 0; i < 40; ++i) {
+ if (read_bvar("h2_stream_context_count") == base_stream_count) {
+ break;
+ }
+ bthread_usleep(5000);
+ }
+ EXPECT_EQ(base_stream_count, read_bvar("h2_stream_context_count"));
+ EXPECT_EQ(base_rst_count + 4, read_bvar("h2_stream_reset_count"));
+}
+
+TEST_F(HttpTest, http2_server_stop_drains_pending_response) {
+ brpc::Server server;
+ brpc::ServerOptions server_options;
+ server_options.idle_timeout_sec = 1;
+ EXPECT_EQ(0, server.AddService(&_svc, brpc::SERVER_DOESNT_OWN_SERVICE));
+ ASSERT_EQ(0, server.Start(0, &server_options));
+
+ butil::fd_guard client_fd(butil::tcp_connect(server.listen_address(),
NULL));
+ ASSERT_GE(client_fd, 0);
+
+ // Advertise a 10-byte initial stream window so the 19-byte JSON response
+ // ({"message":"world"}) sends 10 bytes immediately and leaves 9 bytes
+ // buffered in the server's H2Context after HttpResponseSender destroys the
+ // server-side Controller.
+ brpc::policy::H2Context* cli_ctx =
+ new brpc::policy::H2Context(_h2_client_sock.get(), NULL);
+ ASSERT_EQ(0, cli_ctx->Init());
+ cli_ctx->_client_preface_pending = true;
+ cli_ctx->_unack_local_settings.stream_window_size = 10;
+ cli_ctx->_local_settings.stream_window_size = 10;
+ _h2_client_sock->initialize_parsing_context(&cli_ctx);
+
+ brpc::Controller cntl;
+ cntl.http_request().uri() = "/EchoService/Echo";
+ butil::IOBuf req_out;
+ int h2_stream_id = 0;
+ MakeH2EchoRequestBuf(&req_out, &cntl, &h2_stream_id);
+ const ssize_t req_size = static_cast<ssize_t>(req_out.size());
+ ASSERT_EQ(req_size, req_out.cut_into_file_descriptor(client_fd));
+
+ auto read_bvar = [](const char* name) -> int64_t {
+ std::string val = bvar::Variable::describe_exposed(name);
+ int64_t out = -1;
+ butil::StringToInt64(val, &out);
+ return out;
+ };
+
+ butil::IOPortal resp_buf;
+ for (int i = 0; i < 100; ++i) {
+ int bytes_in_sock = 0;
+ ioctl(client_fd, FIONREAD, &bytes_in_sock);
+ if (bytes_in_sock > 0) {
+ ASSERT_GT(resp_buf.append_from_file_descriptor(client_fd,
bytes_in_sock), 0);
+ brpc::ParseResult pr = brpc::policy::ParseH2Message(
+ &resp_buf, _h2_client_sock.get(), false, NULL);
+ ASSERT_EQ(brpc::PARSE_ERROR_NOT_ENOUGH_DATA, pr.error());
+ }
+ brpc::policy::H2StreamContext* cli_sctx =
cli_ctx->FindStream(h2_stream_id);
+ if (cli_sctx != NULL && cli_sctx->body().size() == 10 &&
+ read_bvar("h2_pending_data_bytes") == 9) {
+ break;
+ }
+ bthread_usleep(5000);
+ }
+ ASSERT_EQ(9, read_bvar("h2_pending_data_bytes"));
+
+ // Stop the server while the remaining 9 bytes of response DATA are waiting
+ // for WINDOW_UPDATE. The connection must stay alive until the deferred
+ // response finishes draining.
+ server.Stop(0);
+ butil::atomic<bool> join_done(false);
+ std::thread join_thread([&]() {
+ server.Join();
+ join_done.store(true, butil::memory_order_release);
+ });
+ bthread_usleep(50000);
+ EXPECT_FALSE(join_done.load(butil::memory_order_acquire));
+ EXPECT_EQ(9, read_bvar("h2_pending_data_bytes"));
+
+ // Send stream and connection WINDOW_UPDATE so the server can flush the
+ // remaining 9 bytes + END_STREAM and finish draining.
+ char winbuf[(brpc::policy::FRAME_HEAD_SIZE + 4) * 2];
+ brpc::policy::SerializeFrameHead(
+ winbuf, 4, brpc::policy::H2_FRAME_WINDOW_UPDATE, 0, h2_stream_id);
+ SaveUint32(winbuf + brpc::policy::FRAME_HEAD_SIZE, 10);
+ brpc::policy::SerializeFrameHead(
+ winbuf + brpc::policy::FRAME_HEAD_SIZE + 4,
+ 4, brpc::policy::H2_FRAME_WINDOW_UPDATE, 0, 0);
+ SaveUint32(winbuf + brpc::policy::FRAME_HEAD_SIZE * 2 + 4, 10);
+ ASSERT_EQ(static_cast<ssize_t>(sizeof(winbuf)),
+ ::write(client_fd, winbuf, sizeof(winbuf)));
+
+ join_thread.join();
+ EXPECT_TRUE(join_done.load(butil::memory_order_acquire));
+ EXPECT_EQ(0, read_bvar("h2_pending_data_bytes"));
+
+ brpc::ParseResult res_pr =
+ brpc::MakeParseError(brpc::PARSE_ERROR_NOT_ENOUGH_DATA);
+ for (int i = 0; i < 100 && !res_pr.is_ok(); ++i) {
+ int bytes_in_sock = 0;
+ ioctl(client_fd, FIONREAD, &bytes_in_sock);
+ if (bytes_in_sock > 0) {
+ ASSERT_GT(resp_buf.append_from_file_descriptor(client_fd,
bytes_in_sock), 0);
+ res_pr = brpc::policy::ParseH2Message(
+ &resp_buf, _h2_client_sock.get(), false, NULL);
+ } else {
+ bthread_usleep(5000);
+ }
+ }
+ ASSERT_TRUE(res_pr.is_ok());
+ ProcessMessage(brpc::policy::ProcessHttpResponse, res_pr.message(), false);
+ ASSERT_FALSE(cntl.Failed()) << cntl.ErrorText();
+ EXPECT_EQ("{\"message\":\"world\"}",
cntl.response_attachment().to_string());
+}
+
+TEST_F(HttpTest, http2_idle_timeout_cleans_up_pending_response) {
+ brpc::Server server;
+ brpc::ServerOptions server_options;
+ server_options.idle_timeout_sec = 1;
+ EXPECT_EQ(0, server.AddService(&_svc, brpc::SERVER_DOESNT_OWN_SERVICE));
+ ASSERT_EQ(0, server.Start(0, &server_options));
+
+ auto read_bvar = [](const char* name) -> int64_t {
+ std::string val = bvar::Variable::describe_exposed(name);
+ int64_t out = -1;
+ butil::StringToInt64(val, &out);
+ return out;
+ };
+
+ auto start_stalled_zero_window_request = [&](butil::fd_guard* out_fd) {
+ butil::fd_guard fd(butil::tcp_connect(server.listen_address(), NULL));
+ ASSERT_GE(fd, 0);
+
+ brpc::policy::H2Context* cli_ctx =
+ new brpc::policy::H2Context(_h2_client_sock.get(), NULL);
+ ASSERT_EQ(0, cli_ctx->Init());
+ cli_ctx->_client_preface_pending = true;
+ cli_ctx->_unack_local_settings.stream_window_size = 0;
+ cli_ctx->_local_settings.stream_window_size = 0;
+ _h2_client_sock->reset_parsing_context(cli_ctx);
+
+ brpc::Controller cntl;
+ cntl.http_request().uri() = "/EchoService/Echo";
+ butil::IOBuf req_out;
+ int h2_stream_id = 0;
+ MakeH2EchoRequestBuf(&req_out, &cntl, &h2_stream_id);
+ const ssize_t req_size = static_cast<ssize_t>(req_out.size());
+ ASSERT_EQ(req_size, req_out.cut_into_file_descriptor(fd));
+
+ butil::IOPortal resp_buf;
+ for (int i = 0; i < 100; ++i) {
+ int bytes_in_sock = 0;
+ ioctl(fd, FIONREAD, &bytes_in_sock);
+ if (bytes_in_sock > 0) {
+ ASSERT_GT(resp_buf.append_from_file_descriptor(fd,
bytes_in_sock), 0);
+ brpc::ParseResult pr = brpc::policy::ParseH2Message(
+ &resp_buf, _h2_client_sock.get(), false, NULL);
+ ASSERT_EQ(brpc::PARSE_ERROR_NOT_ENOUGH_DATA, pr.error());
+ }
+ if (read_bvar("h2_pending_data_bytes") == 19) {
+ break;
+ }
+ bthread_usleep(5000);
+ }
+ ASSERT_EQ(19, read_bvar("h2_pending_data_bytes"));
+ cntl._current_call.stream_user_data->DestroyStreamUserData(
+ _h2_client_sock, &cntl, 0, false);
+ cntl._current_call.stream_user_data = NULL;
+ out_fd->reset(fd.release());
+ };
+
+ // Case 1: While the server is running, idle expiration must fail the
+ // stalled connection and clean up its buffered response even when the peer
+ // keeps TCP open without sending WINDOW_UPDATE or RST_STREAM.
+ butil::fd_guard client_fd1;
+ start_stalled_zero_window_request(&client_fd1);
+ for (int i = 0; i < 300 && read_bvar("h2_pending_data_bytes") != 0; ++i) {
+ bthread_usleep(10000);
+ }
+ ASSERT_EQ(0, read_bvar("h2_pending_data_bytes"));
+
+ // Case 2: If Server::Stop() is called before idle expiration while a
+ // deferred response is buffered on a stalled connection, idle expiration
+ // must still clean up the buffered response and unblock Server::Join().
+ butil::fd_guard client_fd2;
+ start_stalled_zero_window_request(&client_fd2);
+
+ server.Stop(0);
+ butil::atomic<bool> join_done(false);
+ std::thread join_thread([&]() {
+ server.Join();
+ join_done.store(true, butil::memory_order_release);
+ });
+ bthread_usleep(50000);
+ EXPECT_FALSE(join_done.load(butil::memory_order_acquire));
+ EXPECT_EQ(19, read_bvar("h2_pending_data_bytes"));
+
+ for (int i = 0; i < 300 && !join_done.load(butil::memory_order_acquire);
++i) {
+ bthread_usleep(10000);
+ }
+ EXPECT_TRUE(join_done.load(butil::memory_order_acquire));
+ join_thread.join();
+ EXPECT_EQ(0, read_bvar("h2_pending_data_bytes"));
+ EXPECT_EQ(0, read_bvar("h2_unsent_request_count"));
+}
+
+TEST_F(HttpTest,
http2_deferred_response_notify_registration_failure_resets_stream) {
+ brpc::Controller cntl;
+ butil::IOBuf req_out;
+ int h2_stream_id = 0;
+ MakeH2EchoRequestBuf(&req_out, &cntl, &h2_stream_id);
+
+ brpc::policy::H2Context* srv_ctx =
+ new brpc::policy::H2Context(_socket.get(), &_server);
+ ASSERT_EQ(0, srv_ctx->Init());
+ _socket->initialize_parsing_context(&srv_ctx);
+ srv_ctx->_remote_settings.stream_window_size = 0;
+
+ brpc::policy::H2StreamContext* srv_sctx =
+ new brpc::policy::H2StreamContext(false);
+ srv_sctx->Init(srv_ctx, h2_stream_id);
+ srv_sctx->_remote_ended = true;
+ ASSERT_EQ(0, srv_ctx->TryToInsertStream(h2_stream_id, srv_sctx));
+
+ // Saturate _socket->_id_wait_list so Socket::NotifyOnFailed() fails with
+ // EAGAIN while _socket itself remains healthy (!_socket->Failed()).
+ bthread_id_t dummy_id = INVALID_BTHREAD_ID;
+ ASSERT_EQ(0, bthread_id_create(&dummy_id, NULL, NULL));
+ {
+ BAIDU_SCOPED_LOCK(_socket->_id_wait_list_mutex);
+ int consecutive_eagain = 0;
+ while (consecutive_eagain < 26000) {
+ const int rc =
+ bthread_id_list_add(&_socket->_id_wait_list, dummy_id);
+ if (rc == EAGAIN) {
+ ++consecutive_eagain;
+ } else {
+ ASSERT_EQ(0, rc);
+ consecutive_eagain = 0;
+ }
+ }
+ }
+ ASSERT_FALSE(_socket->Failed());
+
+ butil::IOBuf res_out;
+ MakeH2EchoResponseBuf(&res_out, h2_stream_id);
+
+ EXPECT_FALSE(_socket->Failed());
+ EXPECT_EQ(nullptr, srv_ctx->FindStream(h2_stream_id));
+ EXPECT_EQ(0u, srv_ctx->_pending_data_size.load());
+ EXPECT_EQ(0, _socket->_idle_fail_ref_count.load());
+
+ brpc::ParseResult res_pr =
+ brpc::policy::ParseH2Message(&res_out, _h2_client_sock.get(), false,
NULL);
+ ASSERT_TRUE(res_pr.is_ok());
+ ProcessMessage(brpc::policy::ProcessHttpResponse, res_pr.message(), false);
+ ASSERT_TRUE(cntl.Failed());
+ EXPECT_EQ(brpc::EHTTP, cntl.ErrorCode());
+ EXPECT_EQ(brpc::HTTP_STATUS_INTERNAL_SERVER_ERROR,
+ cntl.http_response().status_code());
+ EXPECT_NE(std::string::npos,
+ cntl.ErrorText().find("terminated by RST_STREAM:
INTERNAL_ERROR[0x2]"))
+ << cntl.ErrorText();
+
+ EXPECT_EQ(0, bthread_id_error(dummy_id, 0));
+}
+
TEST_F(HttpTest, http2_settings) {
char settingsbuf[brpc::policy::FRAME_HEAD_SIZE + 36];
brpc::H2Settings h2_settings;
@@ -3292,6 +3700,9 @@ TEST_F(HttpTest, grpc_auth_failed_response) {
// Initialize the stream context with connection context and stream ID
h2_msg->Init(h2_ctx, 1); // stream_id = 1
+ h2_msg->_remote_ended = true;
+ h2_msg->AddOneRefForStage2();
+ ASSERT_EQ(0, h2_ctx->TryToInsertStream(1, h2_msg));
// Set socket and arg using existing test pattern
if (h2_msg->_socket == nullptr) {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
