From e9850e1e52c186ec8936743ea14ebe7296599fa7 Mon Sep 17 00:00:00 2001 From: Wang Xiaofeng Date: Sun, 9 Aug 2026 21:25:32 +0800 Subject: [PATCH 1/2] Support flow-controlled gRPC client requests - Split client request DATA frames according to the peer's connection and stream flow-control windows. - Buffer unsent DATA and resume transmission when WINDOW_UPDATE restores capacity. - Track pending request bytes per H2 connection and apply socket_max_unwritten_bytes as an upper bound. - Reject or reroute new requests once the pending DATA limit is reached. - Release buffered DATA when an RPC fails, times out, or its stream is removed. - Add tests for fragmented transmission, deferred DATA flushing, pending-byte accounting, and buffer cleanup. --- src/brpc/policy/http2_rpc_protocol.cpp | 339 +++++++++++++++-------- src/brpc/policy/http2_rpc_protocol.h | 22 +- test/brpc_grpc_protocol_unittest.cpp | 50 ++++ test/brpc_h2_unsent_message_unittest.cpp | 200 +++++++++++++ test/brpc_http_rpc_protocol_unittest.cpp | 16 +- 5 files changed, 504 insertions(+), 123 deletions(-) diff --git a/src/brpc/policy/http2_rpc_protocol.cpp b/src/brpc/policy/http2_rpc_protocol.cpp index d527a055c2..502ec152e9 100644 --- a/src/brpc/policy/http2_rpc_protocol.cpp +++ b/src/brpc/policy/http2_rpc_protocol.cpp @@ -28,6 +28,7 @@ DECLARE_bool(http_verbose); DECLARE_int32(http_verbose_max_body_length); DECLARE_int32(health_check_interval); DECLARE_bool(usercode_in_pthread); +DECLARE_int64(socket_max_unwritten_bytes); namespace policy { @@ -147,6 +148,12 @@ static int WriteAck(Socket* s, const void* data, size_t n) { return s->Write(&sendbuf, &wopt); } +static int WriteAck(Socket* s, butil::IOBuf* data) { + Socket::WriteOptions wopt; + wopt.ignore_eovercrowded = true; + return s->Write(data, &wopt); +} + // [ https://tools.ietf.org/html/rfc7540#section-6.5.1 ] enum H2SettingsIdentifier { @@ -269,18 +276,13 @@ inline bool AddWindowSize(butil::atomic* window_size, int64_t diff) { // If a sender receives a WINDOW_UPDATE that causes a flow-control window // to exceed this maximum, it MUST terminate either the stream or the connection, // as appropriate. - int64_t before_add = window_size->fetch_add(diff, butil::memory_order_relaxed); - if ((((before_add | diff) >> 31) & 1) == 0) { - // two positive int64_t, check positive overflow - if ((before_add + diff) & (1 << 31)) { - return false; - } - } - if ((((before_add & diff) >> 31) & 1) == 1) { - // two negative int64_t, check negaitive overflow - if (((before_add + diff) & (1 << 31)) == 0) { - return false; - } + const int64_t before_add = + window_size->fetch_add(diff, butil::memory_order_relaxed); + const int64_t after_add = before_add + diff; + if (after_add > std::numeric_limits::max() || + after_add < std::numeric_limits::min()) { + window_size->fetch_sub(diff, butil::memory_order_relaxed); + return false; } // window_size being negative is OK return true; @@ -323,20 +325,19 @@ inline H2Context::FrameHandler FindFrameHandler(H2FrameType type) { H2Context::H2Context(Socket* socket, const Server* server) : _socket(socket) - // Maximize the window size to make sending big request possible before - // receving the remote settings. - , _remote_window_left(H2Settings::MAX_WINDOW_SIZE) + , _remote_window_left(H2Settings::DEFAULT_INITIAL_WINDOW_SIZE) , _conn_state(H2_CONNECTION_UNINITIALIZED) , _last_received_stream_id(-1) , _last_sent_stream_id(1) , _goaway_stream_id(-1) , _remote_settings_received(false) + , _pending_data_size(0) , _deferred_window_update(0) { // Stop printing the field which is useless for remote settings. _remote_settings.connection_window_size = 0; - // Maximize the window size to make sending big request possible before - // receving the remote settings. - _remote_settings.stream_window_size = H2Settings::MAX_WINDOW_SIZE; + // SETTINGS_INITIAL_WINDOW_SIZE defaults to 65535 until the peer sends a + // different value. Larger requests are resumed by WINDOW_UPDATE. + _remote_settings.stream_window_size = H2Settings::DEFAULT_INITIAL_WINDOW_SIZE; if (server) { _unack_local_settings = server->options().h2_settings; } else { @@ -370,6 +371,16 @@ int H2Context::Init() { return 0; } +H2Settings H2Context::remote_settings() const { + std::unique_lock mu(_stream_mutex); + return _remote_settings; +} + +size_t H2Context::VolatilePendingStreamSize() const { + std::unique_lock mu(_stream_mutex); + return _pending_streams.size(); +} + H2StreamContext* H2Context::RemoveStreamAndDeferWU(int stream_id) { H2StreamContext* sctx = NULL; { @@ -377,6 +388,8 @@ H2StreamContext* H2Context::RemoveStreamAndDeferWU(int stream_id) { if (!_pending_streams.erase(stream_id, &sctx)) { return NULL; } + 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 @@ -394,6 +407,7 @@ void H2Context::RemoveGoAwayStreams( std::unique_lock mu(_stream_mutex); _goaway_stream_id = goaway_stream_id; _pending_streams.swap(tmp); + _pending_data_size = 0; } for (StreamMap::const_iterator it = tmp.begin(); it != tmp.end(); ++it) { out_streams->push_back(it->second); @@ -408,6 +422,9 @@ 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()); } } @@ -429,6 +446,9 @@ int H2Context::TryToInsertStream(int stream_id, H2StreamContext* ctx) { } H2StreamContext*& sctx = _pending_streams[stream_id]; if (sctx == NULL) { + // Synchronize creation with SETTINGS_INITIAL_WINDOW_SIZE updates. + ctx->_remote_window_left.store(_remote_settings.stream_window_size, + butil::memory_order_relaxed); sctx = ctx; return 0; } @@ -872,43 +892,27 @@ H2ParseResult H2Context::OnSettings( _local_settings = _unack_local_settings; return MakeH2Message(NULL); } - const int64_t old_stream_window_size = _remote_settings.stream_window_size; - if (!_remote_settings_received) { - // To solve the problem that sender can't send large request before receving - // remote setting, the initial window size of stream/connection is set to - // MAX_WINDOW_SIZE(see constructor of H2Context). - // As a result, in the view of remote side, window size is 65535 by default so - // it may not send its stream size to sender, making stream size still be - // MAX_WINDOW_SIZE. In this case we need to revert this value to default. - H2Settings tmp_settings; - if (!ParseH2Settings(&tmp_settings, it, frame_head.payload_size)) { - LOG(ERROR) << "Fail to parse from SETTINGS"; - return MakeH2Error(H2_PROTOCOL_ERROR); - } - _remote_settings = tmp_settings; - _remote_window_left.fetch_sub( - H2Settings::MAX_WINDOW_SIZE - H2Settings::DEFAULT_INITIAL_WINDOW_SIZE, - butil::memory_order_relaxed); - _remote_settings_received = true; - } else { + int64_t window_diff = 0; + { + std::unique_lock mu(_stream_mutex); + const int64_t old_stream_window_size = + _remote_settings.stream_window_size; if (!ParseH2Settings(&_remote_settings, it, frame_head.payload_size)) { LOG(ERROR) << "Fail to parse from SETTINGS"; return MakeH2Error(H2_PROTOCOL_ERROR); } - } - const int64_t window_diff = - static_cast(_remote_settings.stream_window_size) - - old_stream_window_size; - if (window_diff) { - // Do not update the connection flow-control window here, which can only - // be changed using WINDOW_UPDATE frames. - // https://tools.ietf.org/html/rfc7540#section-6.9.2 - // TODO(gejun): Has race conditions with AppendAndDestroySelf - std::unique_lock mu(_stream_mutex); - for (StreamMap::const_iterator it = _pending_streams.begin(); - it != _pending_streams.end(); ++it) { - if (!AddWindowSize(&it->second->_remote_window_left, window_diff)) { - return MakeH2Error(H2_FLOW_CONTROL_ERROR); + _remote_settings_received = true; + window_diff = static_cast(_remote_settings.stream_window_size) + - old_stream_window_size; + if (window_diff) { + // SETTINGS_INITIAL_WINDOW_SIZE changes all existing stream windows, + // but never the connection-level flow-control window. + for (StreamMap::const_iterator it = _pending_streams.begin(); + it != _pending_streams.end(); ++it) { + if (!AddWindowSize(&it->second->_remote_window_left, + window_diff)) { + return MakeH2Error(H2_FLOW_CONTROL_ERROR); + } } } } @@ -919,6 +923,9 @@ H2ParseResult H2Context::OnSettings( LOG(WARNING) << "Fail to respond settings with ack to " << *_socket; return MakeH2Error(H2_PROTOCOL_ERROR); } + if (window_diff > 0 && !FlushPendingData(0)) { + return MakeH2Error(H2_PROTOCOL_ERROR); + } return MakeH2Message(NULL); } @@ -1026,27 +1033,51 @@ H2ParseResult H2Context::OnWindowUpdate( return MakeH2Error(H2_PROTOCOL_ERROR); } if (frame_head.stream_id == 0) { - if (!AddWindowSize(&_remote_window_left, inc)) { - LOG(ERROR) << "Invalid connection-level window_size_increment=" << inc; - return MakeH2Error(H2_FLOW_CONTROL_ERROR); + { + std::unique_lock mu(_stream_mutex); + if (!AddWindowSize(&_remote_window_left, inc)) { + LOG(ERROR) << "Invalid connection-level window_size_increment=" << inc; + return MakeH2Error(H2_FLOW_CONTROL_ERROR); + } + } + if (!FlushPendingData(0)) { + return MakeH2Error(H2_PROTOCOL_ERROR); } return MakeH2Message(NULL); } else { - H2StreamContext* sctx = FindStream(frame_head.stream_id); - if (sctx == NULL) { - RPC_VLOG << "Fail to find stream_id=" << frame_head.stream_id; - return MakeH2Message(NULL); + { + std::unique_lock mu(_stream_mutex); + H2StreamContext** psctx = _pending_streams.seek(frame_head.stream_id); + if (psctx == nullptr) { + RPC_VLOG << "Fail to find stream_id=" << frame_head.stream_id; + return MakeH2Message(nullptr); + } + if (!AddWindowSize(&(*psctx)->_remote_window_left, inc)) { + 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); + } } - if (!AddWindowSize(&sctx->_remote_window_left, inc)) { - LOG(ERROR) << "Invalid stream-level window_size_increment=" << inc - << " to remote_window_left=" << sctx->_remote_window_left.load(butil::memory_order_relaxed); - return MakeH2Error(H2_FLOW_CONTROL_ERROR); + if (!FlushPendingData(frame_head.stream_id)) { + return MakeH2Error(H2_PROTOCOL_ERROR); } return MakeH2Message(NULL); } } void H2Context::Describe(std::ostream& os, const DescribeOptions& opt) const { + H2Settings remote_settings; + bool remote_settings_received = false; + size_t pending_stream_size = 0; + size_t pending_data_size = 0; + { + std::unique_lock mu(_stream_mutex); + remote_settings = _remote_settings; + remote_settings_received = _remote_settings_received; + pending_stream_size = _pending_streams.size(); + pending_data_size = _pending_data_size; + } if (opt.verbose) { os << '\n'; } @@ -1058,8 +1089,8 @@ void H2Context::Describe(std::ostream& os, const DescribeOptions& opt) const { << _deferred_window_update.load(butil::memory_order_relaxed) << sep << "remote_conn_window_left=" << _remote_window_left.load(butil::memory_order_relaxed) - << sep << "remote_settings=" << _remote_settings - << sep << "remote_settings_received=" << _remote_settings_received + << sep << "remote_settings=" << remote_settings + << sep << "remote_settings_received=" << remote_settings_received << sep << "local_settings=" << _local_settings << sep << "hpacker={"; IndentingOStream os2(os, 2); @@ -1071,7 +1102,8 @@ void H2Context::Describe(std::ostream& os, const DescribeOptions& opt) const { abandoned_size = _abandoned_streams.size(); } os << sep << "abandoned_streams=" << abandoned_size - << sep << "pending_streams=" << VolatilePendingStreamSize(); + << sep << "pending_streams=" << pending_stream_size + << sep << "pending_data_size=" << pending_data_size; if (opt.verbose) { os << '\n'; } @@ -1205,28 +1237,6 @@ void H2StreamContext::SetState(H2StreamState state) { } #endif -bool H2StreamContext::ConsumeWindowSize(int64_t size) { - // This method is guaranteed to be called in AppendAndDestroySelf() which - // is run sequentially. As a result, _remote_window_left of this stream - // context will not be decremented (may be incremented) because following - // AppendAndDestroySelf() are not run yet. - // This fact is important to make window_size changes to stream and - // connection contexts transactionally. - if (_remote_window_left.load(butil::memory_order_relaxed) < size) { - return false; - } - if (!MinusWindowSize(&_conn_ctx->_remote_window_left, size)) { - return false; - } - int64_t after_sub = _remote_window_left.fetch_sub(size, butil::memory_order_relaxed) - size; - if (after_sub < 0) { - LOG(FATAL) << "Impossible, the http2 impl is buggy"; - _remote_window_left.fetch_add(size, butil::memory_order_relaxed); - return false; - } - return true; -} - int H2StreamContext::ConsumeHeaders(butil::IOBufBytesIterator& it) { HPacker& hpacker = _conn_ctx->hpacker(); HttpHeader& h = header(); @@ -1316,43 +1326,53 @@ int H2StreamContext::ConsumeHeaders(butil::IOBufBytesIterator& it) { const CommonStrings* get_common_strings(); -static void PackH2Message(butil::IOBuf* out, +static void PackH2Headers(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(); + uint32_t max_frame_size, + bool end_stream) { char headbuf[FRAME_HEAD_SIZE]; H2FrameHead headers_head = { (uint32_t)headers.size(), H2_FRAME_HEADERS, 0, stream_id}; - if (data.empty() && trailer_headers.empty()) { + if (end_stream) { headers_head.flags |= H2_FLAGS_END_STREAM; } - if (headers_head.payload_size <= remote_settings.max_frame_size) { + if (headers_head.payload_size <= max_frame_size) { headers_head.flags |= H2_FLAGS_END_HEADERS; SerializeFrameHead(headbuf, headers_head); out->append(headbuf, sizeof(headbuf)); out->append(butil::IOBuf::Movable(headers)); } else { - headers_head.payload_size = remote_settings.max_frame_size; + headers_head.payload_size = max_frame_size; SerializeFrameHead(headbuf, headers_head); out->append(headbuf, sizeof(headbuf)); headers.cutn(out, headers_head.payload_size); H2FrameHead cont_head = {0, H2_FRAME_CONTINUATION, 0, stream_id}; while (!headers.empty()) { - if (headers.size() <= remote_settings.max_frame_size) { + if (headers.size() <= max_frame_size) { cont_head.flags |= H2_FLAGS_END_HEADERS; cont_head.payload_size = headers.size(); } else { - cont_head.payload_size = remote_settings.max_frame_size; + cont_head.payload_size = max_frame_size; } SerializeFrameHead(headbuf, cont_head); out->append(headbuf, FRAME_HEAD_SIZE); headers.cutn(out, cont_head.payload_size); } } +} + +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); @@ -1388,6 +1408,92 @@ static void PackH2Message(butil::IOBuf* out, } } +void H2Context::AppendPendingDataLocked(H2StreamContext* sctx, + butil::IOBuf* out) { + CHECK(sctx != nullptr); + const uint32_t max_frame_size = _remote_settings.max_frame_size; + char headbuf[FRAME_HEAD_SIZE]; + while (!sctx->_pending_data.empty()) { + 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); + if (conn_window <= 0 || stream_window <= 0) { + break; + } + const size_t payload_size = std::min( + sctx->_pending_data.size(), + std::min(static_cast(max_frame_size), + static_cast(std::min(conn_window, stream_window)))); + CHECK_GT(payload_size, 0u); + _remote_window_left.fetch_sub(payload_size, butil::memory_order_relaxed); + sctx->_remote_window_left.fetch_sub(payload_size, + butil::memory_order_relaxed); + + H2FrameHead data_head = { + static_cast(payload_size), H2_FRAME_DATA, 0, + sctx->stream_id()}; + if (payload_size == sctx->_pending_data.size()) { + 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; + } +} + +void H2Context::AppendClientRequestData(H2StreamContext* sctx, + const butil::IOBuf& data, + butil::IOBuf* out) { + std::unique_lock mu(_stream_mutex); + CHECK(sctx->_pending_data.empty()); + sctx->_pending_data = data; + _pending_data_size += data.size(); + AppendPendingDataLocked(sctx, out); +} + +void H2Context::ClearPendingData(int stream_id) { + std::unique_lock mu(_stream_mutex); + H2StreamContext** psctx = _pending_streams.seek(stream_id); + if (psctx == nullptr) { + return; + } + CHECK_GE(_pending_data_size, (*psctx)->_pending_data.size()); + _pending_data_size -= (*psctx)->_pending_data.size(); + (*psctx)->_pending_data.clear(); +} + +bool H2Context::PendingDataOvercrowded() const { + std::unique_lock mu(_stream_mutex); + return FLAGS_socket_max_unwritten_bytes > 0 && + _pending_data_size >= + static_cast(FLAGS_socket_max_unwritten_bytes); +} + +bool H2Context::FlushPendingData(int stream_id) { + butil::IOBuf out; + { + std::unique_lock mu(_stream_mutex); + if (stream_id != 0) { + H2StreamContext** psctx = _pending_streams.seek(stream_id); + if (psctx != nullptr) { + AppendPendingDataLocked(*psctx, &out); + } + } else { + for (StreamMap::const_iterator it = _pending_streams.begin(); + it != _pending_streams.end(); ++it) { + if (_remote_window_left.load(butil::memory_order_relaxed) <= 0) { + break; + } + AppendPendingDataLocked(it->second, &out); + } + } + } + return out.empty() || WriteAck(_socket, &out) == 0; +} + H2UnsentRequest* H2UnsentRequest::New(Controller* c) { const HttpHeader& h = c->http_request(); const CommonStrings* const common = get_common_strings(); @@ -1485,12 +1591,13 @@ void H2UnsentRequest::DestroyStreamUserData(SocketUniquePtr& sending_sock, int error_code, bool /*end_of_rpc*/) { RemoveRefOnQuit deref_self(this); - if (sending_sock != NULL && error_code != 0) { + if (sending_sock != nullptr && error_code != 0) { CHECK_EQ(cntl, _cntl); std::unique_lock mu(_mutex); _cntl = NULL; if (_stream_id != 0) { H2Context* ctx = static_cast(sending_sock->parsing_context()); + ctx->ClearPendingData(_stream_id); ctx->AddAbandonedStream(_stream_id); } } @@ -1538,6 +1645,10 @@ H2UnsentRequest::AppendAndDestroySelf(butil::IOBuf* out, Socket* socket) { if (ctx->VolatilePendingStreamSize() > ctx->remote_settings().max_concurrent_streams) { return butil::Status(ELIMIT, "Pending Stream count exceeds max concurrent stream"); } + if (ctx->PendingDataOvercrowded()) { + return butil::Status(EOVERCROWDED, + "Too much pending HTTP/2 request data"); + } // Although the critical section looks huge, it should rarely be contended // since timeout of RPC is much larger than the delay of sending. @@ -1557,21 +1668,14 @@ H2UnsentRequest::AppendAndDestroySelf(butil::IOBuf* out, Socket* socket) { } _sctx->Init(ctx, id); - // check flow control restriction - if (!_cntl->request_attachment().empty()) { - const int64_t data_size = _cntl->request_attachment().size(); - if (!_sctx->ConsumeWindowSize(data_size)) { - return butil::Status(ELIMIT, "remote_window_left is not enough, data_size=%" PRId64, data_size); - } - } - const int rc = ctx->TryToInsertStream(id, _sctx.get()); if (rc < 0) { return butil::Status(EINTERNAL, "Fail to insert existing stream_id"); } else if (rc > 0) { return butil::Status(ELOGOFF, "the connection just issued GOAWAY"); } - _stream_id = _sctx->stream_id(); + H2StreamContext* const sctx = _sctx.get(); + _stream_id = sctx->stream_id(); // After calling TryToInsertStream, the ownership of _sctx is transferred to ctx _sctx.release(); @@ -1580,7 +1684,8 @@ H2UnsentRequest::AppendAndDestroySelf(butil::IOBuf* out, Socket* socket) { 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) { + const H2Settings remote_settings = ctx->remote_settings(); + if (remote_settings.header_table_size == 0) { options.index_policy = HPACK_NEVER_INDEX_HEADER; } @@ -1597,8 +1702,19 @@ H2UnsentRequest::AppendAndDestroySelf(butil::IOBuf* out, Socket* socket) { } butil::IOBuf frag; appender.move_to(frag); - butil::IOBuf dummy_buf; - PackH2Message(out, frag, dummy_buf, _cntl->request_attachment(), _stream_id, ctx); + const butil::IOBuf& request_data = _cntl->request_attachment(); + PackH2Headers(out, frag, _stream_id, remote_settings.max_frame_size, + request_data.empty()); + if (!request_data.empty()) { + ctx->AppendClientRequestData(sctx, request_data, out); + } + 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)); + } return butil::Status::OK(); } @@ -1821,7 +1937,8 @@ void PackH2Request(butil::IOBuf*, static bool IsH2SocketValid(Socket* s) { H2Context* c = static_cast(s->parsing_context()); - return (c == NULL || !c->RunOutStreams()); + return c == nullptr || + (!c->RunOutStreams() && !c->PendingDataOvercrowded()); } StreamUserData* H2GlobalStreamCreator::OnCreatingStream( diff --git a/src/brpc/policy/http2_rpc_protocol.h b/src/brpc/policy/http2_rpc_protocol.h index b4422ee057..540c75e245 100644 --- a/src/brpc/policy/http2_rpc_protocol.h +++ b/src/brpc/policy/http2_rpc_protocol.h @@ -258,8 +258,6 @@ class H2StreamContext : public HttpContext { return _deferred_window_update.exchange(0, butil::memory_order_relaxed); } - bool ConsumeWindowSize(int64_t size); - #if defined(BRPC_H2_STREAM_STATE) H2StreamState state() const { return _state; } void SetState(H2StreamState state); @@ -276,6 +274,9 @@ friend class H2Context; butil::atomic _deferred_window_update; uint64_t _correlation_id; butil::IOBuf _remaining_header_fragment; + // Request body which cannot be sent yet due to remote flow control. + // Accessed under H2Context::_stream_mutex. + butil::IOBuf _pending_data; }; StreamCreator* get_h2_global_stream_creator(); @@ -319,7 +320,7 @@ class H2Context : public Destroyable, public Describable { // main_socket: the socket owns this object as parsing_context // server: NULL means client-side H2Context(Socket* main_socket, const Server* server); - ~H2Context(); + ~H2Context() override; // Must be called before usage. int Init(); @@ -337,10 +338,13 @@ class H2Context : public Destroyable, public Describable { // Try to map stream_id to ctx if stream_id does not exist before // Returns 0 on success, -1 on exist, 1 on goaway. int TryToInsertStream(int stream_id, H2StreamContext* ctx); - size_t VolatilePendingStreamSize() const { return _pending_streams.size(); } + size_t VolatilePendingStreamSize() const; + bool PendingDataOvercrowded() const; HPacker& hpacker() { return _hpacker; } - const H2Settings& remote_settings() const { return _remote_settings; } + // Return a consistent snapshot because SETTINGS may be processed by the + // socket reader while a request is being packed by a writer. + H2Settings remote_settings() const; const H2Settings& local_settings() const { return _local_settings; } bool is_client_side() const { return _socket->CreatedByConnect(); } @@ -374,6 +378,11 @@ friend void InitFrameHandlers(); void RemoveGoAwayStreams(int goaway_stream_id, std::vector* out_streams); H2StreamContext* FindStream(int stream_id); + void AppendClientRequestData(H2StreamContext*, const butil::IOBuf&, + butil::IOBuf*); + void AppendPendingDataLocked(H2StreamContext*, butil::IOBuf*); + void ClearPendingData(int stream_id); + bool FlushPendingData(int stream_id); // True if the connection is established by client, otherwise it's // accepted by server. @@ -393,6 +402,9 @@ friend void InitFrameHandlers(); typedef butil::FlatMap StreamMap; mutable butil::Mutex _stream_mutex; StreamMap _pending_streams; + // Total bytes retained in H2StreamContext::_pending_data on this + // connection. Accessed under _stream_mutex. + size_t _pending_data_size; butil::atomic _deferred_window_update; }; diff --git a/test/brpc_grpc_protocol_unittest.cpp b/test/brpc_grpc_protocol_unittest.cpp index f170639d17..eb1b988e88 100644 --- a/test/brpc_grpc_protocol_unittest.cpp +++ b/test/brpc_grpc_protocol_unittest.cpp @@ -88,6 +88,17 @@ class MyGrpcService : public ::test::GrpcService { } }; +class WindowGrpcService : public ::test::GrpcService { +public: + void Method(::google::protobuf::RpcController*, + const ::test::GrpcRequest* req, + ::test::GrpcResponse* res, + ::google::protobuf::Closure* done) override { + brpc::ClosureGuard done_guard(done); + res->set_message(req->message()); + } +}; + class GrpcTest : public ::testing::Test { protected: GrpcTest() { @@ -272,4 +283,43 @@ TEST_F(GrpcTest, GrpcTimeOut) { } } +TEST(GrpcProtocol, client_sends_large_request_with_small_remote_window) { + WindowGrpcService service; + brpc::Server server; + ASSERT_EQ(0, server.AddService(&service, brpc::SERVER_DOESNT_OWN_SERVICE)); + brpc::ServerOptions server_options; + server_options.h2_settings.stream_window_size = 32; + ASSERT_EQ(0, server.Start("127.0.0.1:8012", &server_options)); + + brpc::Channel channel; + brpc::ChannelOptions channel_options; + channel_options.protocol = g_protocol; + channel_options.timeout_ms = 10000; + ASSERT_EQ(0, channel.Init("127.0.0.1:8012", "", &channel_options)); + test::GrpcService_Stub stub(&channel); + + // Establish the H2 connection and receive the server SETTINGS first. + { + 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(request.message(), response.message()); + } + + test::GrpcRequest request; + test::GrpcResponse response; + brpc::Controller cntl; + request.set_message(std::string(128 * 1024, 'x')); + request.set_gzip(false); + request.set_return_error(false); + stub.Method(&cntl, &request, &response, nullptr); + EXPECT_FALSE(cntl.Failed()) << cntl.ErrorText(); + EXPECT_EQ(request.message(), response.message()); +} + } // namespace diff --git a/test/brpc_h2_unsent_message_unittest.cpp b/test/brpc_h2_unsent_message_unittest.cpp index 5e3b266dfe..3b36599840 100644 --- a/test/brpc_h2_unsent_message_unittest.cpp +++ b/test/brpc_h2_unsent_message_unittest.cpp @@ -27,11 +27,211 @@ #include "brpc/policy/http2_rpc_protocol.h" #include "gperftools_helper.h" +namespace brpc { +DECLARE_int64(socket_max_unwritten_bytes); +} + int main(int argc, char* argv[]) { testing::InitGoogleTest(&argc, argv); return RUN_ALL_TESTS(); } +namespace { + +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))); + brpc::policy::H2FrameHead frame; + frame.payload_size = + (static_cast(head[0]) << 16) | + (static_cast(head[1]) << 8) | + static_cast(head[2]); + frame.type = static_cast(head[3]); + frame.flags = head[4]; + frame.stream_id = + (static_cast(head[5]) << 24) | + (static_cast(head[6]) << 16) | + (static_cast(head[7]) << 8) | + static_cast(head[8]); + payload->resize(frame.payload_size); + CHECK_EQ(frame.payload_size, + buf->cutn(&(*payload)[0], frame.payload_size)); + return frame; +} + +} // namespace + +TEST(H2UnsentMessage, split_request_data_by_remote_window) { + brpc::SocketId id; + brpc::SocketUniquePtr sock; + brpc::SocketOptions options; + options.user = brpc::get_client_side_messenger(); + 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(), nullptr); + 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)); + + butil::IOBuf body; + body.append("abcdefghij"); + butil::IOBuf out; + ctx->AppendClientRequestData(sctx, body, &out); + + std::string payload; + brpc::policy::H2FrameHead frame = PopFrame(&out, &payload); + EXPECT_EQ(4u, frame.payload_size); + EXPECT_EQ(brpc::policy::H2_FRAME_DATA, frame.type); + EXPECT_EQ(0, frame.flags & 0x1); + EXPECT_EQ("abcd", payload); + frame = PopFrame(&out, &payload); + EXPECT_EQ(1u, frame.payload_size); + EXPECT_EQ(0, frame.flags & 0x1); + EXPECT_EQ("e", payload); + EXPECT_TRUE(out.empty()); + EXPECT_EQ(5u, sctx->_pending_data.size()); + EXPECT_EQ(5u, ctx->_pending_data_size); + EXPECT_EQ(1, ctx->_remote_window_left); + EXPECT_EQ(0, sctx->_remote_window_left); + + 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 mu(ctx->_stream_mutex); + ctx->AppendPendingDataLocked(sctx, &out); + } + frame = PopFrame(&out, &payload); + EXPECT_EQ(3u, frame.payload_size); + EXPECT_EQ(0, frame.flags & 0x1); + EXPECT_EQ("fgh", payload); + EXPECT_TRUE(out.empty()); + EXPECT_EQ(2u, ctx->_pending_data_size); + + 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 mu(ctx->_stream_mutex); + ctx->AppendPendingDataLocked(sctx, &out); + } + frame = PopFrame(&out, &payload); + EXPECT_EQ(2u, frame.payload_size); + EXPECT_NE(0, frame.flags & 0x1); + EXPECT_EQ("ij", payload); + EXPECT_TRUE(out.empty()); + EXPECT_TRUE(sctx->_pending_data.empty()); + EXPECT_EQ(0u, ctx->_pending_data_size); +} + +TEST(H2UnsentMessage, request_does_not_fail_when_body_exceeds_window) { + brpc::SocketId id; + brpc::SocketUniquePtr sock; + brpc::SocketOptions options; + options.user = brpc::get_client_side_messenger(); + 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(), nullptr); + ASSERT_EQ(0, ctx->Init()); + sock->initialize_parsing_context(&ctx); + ctx->_last_sent_stream_id = 1; + ctx->_remote_settings.max_frame_size = 4; + ctx->_remote_settings.stream_window_size = 3; + ctx->_remote_window_left = 3; + + brpc::Controller cntl; + cntl.http_request().uri() = "http://example.com/echo"; + cntl.request_attachment().append("abcdefghij"); + brpc::policy::H2UnsentRequest* request = + brpc::policy::H2UnsentRequest::New(&cntl); + ASSERT_TRUE(request != nullptr); + + butil::IOBuf out; + const butil::Status status = + request->AppendAndDestroySelf(&out, sock.get()); + EXPECT_TRUE(status.ok()) << status; + 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(0, ctx->_remote_window_left); + EXPECT_EQ(0, sctx->_remote_window_left); +} + +TEST(H2UnsentMessage, invalid_window_update_does_not_change_window) { + brpc::SocketId id; + brpc::SocketUniquePtr sock; + brpc::SocketOptions options; + options.user = brpc::get_client_side_messenger(); + 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(), nullptr); + ASSERT_EQ(0, ctx->Init()); + sock->initialize_parsing_context(&ctx); + const int64_t max_window_size = std::numeric_limits::max(); + 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)); +} + +TEST(H2UnsentMessage, clear_pending_data_releases_overcrowded_buffer) { + brpc::SocketId id; + brpc::SocketUniquePtr sock; + brpc::SocketOptions options; + options.user = brpc::get_client_side_messenger(); + 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(), nullptr); + 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)); + + butil::IOBuf body; + body.append("pending"); + butil::IOBuf out; + ctx->AppendClientRequestData(sctx, body, &out); + ASSERT_TRUE(out.empty()); + ASSERT_EQ(body.size(), ctx->_pending_data_size); + + const int64_t saved_limit = brpc::FLAGS_socket_max_unwritten_bytes; + brpc::FLAGS_socket_max_unwritten_bytes = body.size(); + EXPECT_TRUE(ctx->PendingDataOvercrowded()); + ctx->ClearPendingData(1); + EXPECT_FALSE(ctx->PendingDataOvercrowded()); + brpc::FLAGS_socket_max_unwritten_bytes = saved_limit; + + EXPECT_TRUE(sctx->_pending_data.empty()); + EXPECT_EQ(0u, ctx->_pending_data_size); +} + TEST(H2UnsentMessage, request_throughput) { brpc::Controller cntl; butil::IOBuf request_buf; diff --git a/test/brpc_http_rpc_protocol_unittest.cpp b/test/brpc_http_rpc_protocol_unittest.cpp index 87837abf1a..9c703195e8 100644 --- a/test/brpc_http_rpc_protocol_unittest.cpp +++ b/test/brpc_http_rpc_protocol_unittest.cpp @@ -1527,7 +1527,7 @@ TEST_F(HttpTest, http2_rst_after_header_and_data) { ASSERT_TRUE(cntl.http_response().status_code() == brpc::HTTP_STATUS_OK); } -TEST_F(HttpTest, http2_window_used_up) { +TEST_F(HttpTest, http2_window_used_up_buffers_request) { brpc::Controller cntl; butil::IOBuf request_buf; test::EchoRequest req; @@ -1545,6 +1545,8 @@ TEST_F(HttpTest, http2_window_used_up) { buf.append(settingsbuf, brpc::policy::FRAME_HEAD_SIZE + nb); brpc::policy::ParseH2Message(&buf, _h2_client_sock.get(), false, NULL); + brpc::policy::H2Context* ctx = static_cast( + _h2_client_sock->parsing_context()); int nsuc = brpc::H2Settings::DEFAULT_INITIAL_WINDOW_SIZE / cntl.request_attachment().size(); for (int i = 0; i <= nsuc; i++) { brpc::policy::H2UnsentRequest* h2_req = brpc::policy::H2UnsentRequest::New(&cntl); @@ -1554,15 +1556,15 @@ TEST_F(HttpTest, http2_window_used_up) { NULL, &cntl, request_buf, NULL); butil::IOBuf dummy; butil::Status st = socket_message->AppendAndDestroySelf(&dummy, _h2_client_sock.get()); + ASSERT_TRUE(st.ok()); if (i == nsuc) { - // the last message should fail according to flow control policy. - ASSERT_FALSE(st.ok()); - ASSERT_TRUE(st.error_code() == brpc::ELIMIT); - ASSERT_TRUE(butil::StringPiece(st.error_str()).starts_with("remote_window_left is not enough")); + ASSERT_GT(ctx->_pending_data_size, 0u); + h2_req->DestroyStreamUserData( + _h2_client_sock, &cntl, ECANCELED, false); + ASSERT_EQ(0u, ctx->_pending_data_size); } else { - ASSERT_TRUE(st.ok()); + h2_req->DestroyStreamUserData(_h2_client_sock, &cntl, 0, false); } - h2_req->DestroyStreamUserData(_h2_client_sock, &cntl, 0, false); } } From f55a7a4abbfb86803643483739d40f8f283f57ce Mon Sep 17 00:00:00 2001 From: Wang Xiaofeng Date: Mon, 10 Aug 2026 23:47:51 +0800 Subject: [PATCH 2/2] Fix pending HTTP/2 data limit race - Check pending DATA capacity atomically with client stream insertion. - Leave stream and window state unchanged when the limit is exceeded. - Use an ephemeral port in the gRPC flow-control test. - Avoid accessing an empty payload buffer in H2 frame tests. --- src/brpc/policy/http2_rpc_protocol.cpp | 86 ++++++++++++++++-------- src/brpc/policy/http2_rpc_protocol.h | 8 ++- test/brpc_grpc_protocol_unittest.cpp | 4 +- test/brpc_h2_unsent_message_unittest.cpp | 42 +++++++----- test/brpc_http_rpc_protocol_unittest.cpp | 6 +- 5 files changed, 96 insertions(+), 50 deletions(-) diff --git a/src/brpc/policy/http2_rpc_protocol.cpp b/src/brpc/policy/http2_rpc_protocol.cpp index 502ec152e9..eb0e35317d 100644 --- a/src/brpc/policy/http2_rpc_protocol.cpp +++ b/src/brpc/policy/http2_rpc_protocol.cpp @@ -1444,14 +1444,57 @@ void H2Context::AppendPendingDataLocked(H2StreamContext* sctx, } } -void H2Context::AppendClientRequestData(H2StreamContext* sctx, - const butil::IOBuf& data, - butil::IOBuf* out) { +butil::Status H2Context::TryToInsertClientStream( + int stream_id, H2StreamContext* sctx, const butil::IOBuf& data, + butil::IOBuf* out) { std::unique_lock mu(_stream_mutex); - CHECK(sctx->_pending_data.empty()); - sctx->_pending_data = data; - _pending_data_size += data.size(); - AppendPendingDataLocked(sctx, out); + if (_goaway_stream_id >= 0 && stream_id > _goaway_stream_id) { + return butil::Status(ELOGOFF, "the connection just issued GOAWAY"); + } + if (_pending_streams.seek(stream_id) != nullptr) { + return butil::Status(EINTERNAL, + "Fail to insert existing stream_id"); + } + if (_pending_streams.size() >= _remote_settings.max_concurrent_streams) { + return butil::Status( + 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); + size_t sendable_size = 0; + if (conn_window > 0 && stream_window > 0) { + sendable_size = std::min( + data.size(), + static_cast(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(FLAGS_socket_max_unwritten_bytes); + // 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) { + return butil::Status(EOVERCROWDED, + "Too much pending HTTP/2 request data"); + } + } + + // Mutate stream and window state only after all failure checks above. + _pending_streams[stream_id] = sctx; + if (!data.empty()) { + CHECK(sctx->_pending_data.empty()); + sctx->_pending_data = data; + _pending_data_size += data.size(); + AppendPendingDataLocked(sctx, out); + } + return butil::Status::OK(); } void H2Context::ClearPendingData(int stream_id) { @@ -1641,15 +1684,6 @@ H2UnsentRequest::AppendAndDestroySelf(butil::IOBuf* out, Socket* socket) { out->append(settingsbuf, nb); } - // TODO(zhujiashun): also check this in server push - if (ctx->VolatilePendingStreamSize() > ctx->remote_settings().max_concurrent_streams) { - return butil::Status(ELIMIT, "Pending Stream count exceeds max concurrent stream"); - } - if (ctx->PendingDataOvercrowded()) { - return butil::Status(EOVERCROWDED, - "Too much pending HTTP/2 request data"); - } - // 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 mu(_mutex); @@ -1668,16 +1702,7 @@ H2UnsentRequest::AppendAndDestroySelf(butil::IOBuf* out, Socket* socket) { } _sctx->Init(ctx, id); - const int rc = ctx->TryToInsertStream(id, _sctx.get()); - if (rc < 0) { - return butil::Status(EINTERNAL, "Fail to insert existing stream_id"); - } else if (rc > 0) { - return butil::Status(ELOGOFF, "the connection just issued GOAWAY"); - } H2StreamContext* const sctx = _sctx.get(); - _stream_id = sctx->stream_id(); - // After calling TryToInsertStream, the ownership of _sctx is transferred to ctx - _sctx.release(); HPacker& hpacker = ctx->hpacker(); butil::IOBufAppender appender; @@ -1703,11 +1728,16 @@ H2UnsentRequest::AppendAndDestroySelf(butil::IOBuf* out, Socket* socket) { butil::IOBuf frag; appender.move_to(frag); const butil::IOBuf& request_data = _cntl->request_attachment(); - PackH2Headers(out, frag, _stream_id, remote_settings.max_frame_size, + PackH2Headers(out, frag, id, remote_settings.max_frame_size, request_data.empty()); - if (!request_data.empty()) { - ctx->AppendClientRequestData(sctx, request_data, out); + const butil::Status insert_status = + ctx->TryToInsertClientStream(id, sctx, request_data, out); + if (!insert_status.ok()) { + return insert_status; } + _stream_id = id; + // TryToInsertClientStream transfers ownership of _sctx to ctx on success. + _sctx.release(); const int64_t conn_wu = ctx->ReleaseDeferredWindowUpdate(); if (conn_wu > 0) { char winbuf[FRAME_HEAD_SIZE + 4]; diff --git a/src/brpc/policy/http2_rpc_protocol.h b/src/brpc/policy/http2_rpc_protocol.h index 540c75e245..27055ae9a5 100644 --- a/src/brpc/policy/http2_rpc_protocol.h +++ b/src/brpc/policy/http2_rpc_protocol.h @@ -378,8 +378,12 @@ friend void InitFrameHandlers(); void RemoveGoAwayStreams(int goaway_stream_id, std::vector* out_streams); H2StreamContext* FindStream(int stream_id); - void AppendClientRequestData(H2StreamContext*, const butil::IOBuf&, - butil::IOBuf*); + // 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*); void AppendPendingDataLocked(H2StreamContext*, butil::IOBuf*); void ClearPendingData(int stream_id); bool FlushPendingData(int stream_id); diff --git a/test/brpc_grpc_protocol_unittest.cpp b/test/brpc_grpc_protocol_unittest.cpp index eb1b988e88..5a9752ea23 100644 --- a/test/brpc_grpc_protocol_unittest.cpp +++ b/test/brpc_grpc_protocol_unittest.cpp @@ -289,13 +289,13 @@ TEST(GrpcProtocol, client_sends_large_request_with_small_remote_window) { ASSERT_EQ(0, server.AddService(&service, brpc::SERVER_DOESNT_OWN_SERVICE)); brpc::ServerOptions server_options; server_options.h2_settings.stream_window_size = 32; - ASSERT_EQ(0, server.Start("127.0.0.1:8012", &server_options)); + ASSERT_EQ(0, server.Start("127.0.0.1:0", &server_options)); brpc::Channel channel; brpc::ChannelOptions channel_options; channel_options.protocol = g_protocol; channel_options.timeout_ms = 10000; - ASSERT_EQ(0, channel.Init("127.0.0.1:8012", "", &channel_options)); + ASSERT_EQ(0, channel.Init(server.listen_address(), &channel_options)); test::GrpcService_Stub stub(&channel); // Establish the H2 connection and receive the server SETTINGS first. diff --git a/test/brpc_h2_unsent_message_unittest.cpp b/test/brpc_h2_unsent_message_unittest.cpp index 3b36599840..1c7b985aed 100644 --- a/test/brpc_h2_unsent_message_unittest.cpp +++ b/test/brpc_h2_unsent_message_unittest.cpp @@ -54,8 +54,10 @@ brpc::policy::H2FrameHead PopFrame(butil::IOBuf* buf, std::string* payload) { (static_cast(head[7]) << 8) | static_cast(head[8]); payload->resize(frame.payload_size); - CHECK_EQ(frame.payload_size, - buf->cutn(&(*payload)[0], frame.payload_size)); + if (frame.payload_size != 0) { + CHECK_EQ(frame.payload_size, + buf->cutn(&(*payload)[0], frame.payload_size)); + } return frame; } @@ -80,12 +82,11 @@ TEST(H2UnsentMessage, split_request_data_by_remote_window) { brpc::policy::H2StreamContext* sctx = new brpc::policy::H2StreamContext(false); sctx->Init(ctx, 1); - ASSERT_EQ(0, ctx->TryToInsertStream(1, sctx)); butil::IOBuf body; body.append("abcdefghij"); butil::IOBuf out; - ctx->AppendClientRequestData(sctx, body, &out); + ASSERT_TRUE(ctx->TryToInsertClientStream(1, sctx, body, &out).ok()); std::string payload; brpc::policy::H2FrameHead frame = PopFrame(&out, &payload); @@ -209,26 +210,37 @@ TEST(H2UnsentMessage, clear_pending_data_releases_overcrowded_buffer) { 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)); - butil::IOBuf body; body.append("pending"); butil::IOBuf out; - ctx->AppendClientRequestData(sctx, body, &out); + GFLAGS_NAMESPACE::FlagSaver flag_saver; + brpc::FLAGS_socket_max_unwritten_bytes = body.size(); + + std::unique_ptr sctx( + new brpc::policy::H2StreamContext(false)); + sctx->Init(ctx, 1); + ASSERT_TRUE( + 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); - - const int64_t saved_limit = brpc::FLAGS_socket_max_unwritten_bytes; - brpc::FLAGS_socket_max_unwritten_bytes = body.size(); EXPECT_TRUE(ctx->PendingDataOvercrowded()); + + std::unique_ptr rejected_sctx( + new brpc::policy::H2StreamContext(false)); + rejected_sctx->Init(ctx, 3); + butil::IOBuf rejected_body; + rejected_body.append("x"); + const butil::Status rejected = ctx->TryToInsertClientStream( + 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); + ctx->ClearPendingData(1); EXPECT_FALSE(ctx->PendingDataOvercrowded()); - brpc::FLAGS_socket_max_unwritten_bytes = saved_limit; - EXPECT_TRUE(sctx->_pending_data.empty()); + EXPECT_TRUE(inserted_sctx->_pending_data.empty()); EXPECT_EQ(0u, ctx->_pending_data_size); } diff --git a/test/brpc_http_rpc_protocol_unittest.cpp b/test/brpc_http_rpc_protocol_unittest.cpp index 9c703195e8..97de699547 100644 --- a/test/brpc_http_rpc_protocol_unittest.cpp +++ b/test/brpc_http_rpc_protocol_unittest.cpp @@ -1408,12 +1408,12 @@ TEST_F(HttpTest, http2_sanity) { options.protocol = "h2"; ASSERT_EQ(0, channel.Init(butil::EndPoint(butil::my_ip(), port), &options)); - // Check that the first request with size larger than the default window can - // be sent out, when remote settings are not received. + // Check that the first request larger than the default window completes + // after SETTINGS and WINDOW_UPDATE make more capacity available. brpc::Controller cntl; test::EchoRequest big_req; test::EchoResponse res; - std::string message(2 * 1024 * 1024 /* 2M */, 'x'); + std::string message(128 * 1024, 'x'); big_req.set_message(message); cntl.http_request().set_method(brpc::HTTP_METHOD_POST); cntl.http_request().uri() = "/EchoService/Echo";