You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@brpc.apache.org by "jenrryyou (via GitHub)" <gi...@apache.org> on 2026/02/05 07:52:35 UTC
[PR] Make batch create stream SendFeedback thread safe (brpc)
jenrryyou opened a new pull request, #3215:
URL: https://github.com/apache/brpc/pull/3215
### What problem does this PR solve?
Issue Number: resolve
Fixes https://github.com/apache/brpc/pull/2754
Problem Summary:
批量创建stream场景下，解决 Stream SetConnected() 与 Consume() 并发导致的 SendFeedback
发送与已消费字节统计的竞态问题，使得反馈的已消费字节数准确，避免发送端错误反压。
### What is changed and the side effects?
Changed:
Side effects:
- Performance effects:
1. 对非批量创建场景，会走到fast path，对性能无任何影响；
2. 对批量创建场景，连接建立前的消费路径会略增 CPU，但只发生在“建连窗口期”，创建完后对性能无任何影响。
- Breaking backward compatibility:
兼容
---
### Check List:
- Please make sure your changes are compilable.
- When providing us with a new feature, it is best to add related tests.
- Please follow [Contributor Covenant Code of Conduct](https://github.com/apache/brpc/blob/master/CODE_OF_CONDUCT.md).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@brpc.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@brpc.apache.org
For additional commands, e-mail: dev-help@brpc.apache.org
Re: [PR] Make batch create stream SendFeedback thread safe (brpc)
Posted by "wwbmmm (via GitHub)" <gi...@apache.org>.
wwbmmm merged PR #3215:
URL: https://github.com/apache/brpc/pull/3215
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@brpc.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@brpc.apache.org
For additional commands, e-mail: dev-help@brpc.apache.org
Re: [PR] Make batch create stream SendFeedback thread safe (brpc)
Posted by "jenrryyou (via GitHub)" <gi...@apache.org>.
jenrryyou commented on PR #3215:
URL: https://github.com/apache/brpc/pull/3215#issuecomment-4043687765
@wwbmmm hi, I've modified the RP according to copilot's comments.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@brpc.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@brpc.apache.org
For additional commands, e-mail: dev-help@brpc.apache.org
Re: [PR] Make batch create stream SendFeedback thread safe (brpc)
Posted by "wwbmmm (via GitHub)" <gi...@apache.org>.
wwbmmm commented on PR #3215:
URL: https://github.com/apache/brpc/pull/3215#issuecomment-3857467195
Can you add some unit test to verify this scenario?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@brpc.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@brpc.apache.org
For additional commands, e-mail: dev-help@brpc.apache.org
Re: [PR] Make batch create stream SendFeedback thread safe (brpc)
Posted by "jenrryyou (via GitHub)" <gi...@apache.org>.
jenrryyou commented on PR #3215:
URL: https://github.com/apache/brpc/pull/3215#issuecomment-4003505399
> Can you add some unit test to verify this scenario?
• 新增用例：test/brpc_streaming_rpc_unittest.cpp（StreamingRpcTest.batch_create_stream_feedback_race）
• 核心思路：客户端批量创建 2 条 request stream，刻意阻塞 extra stream 的 SetConnected()（锁住其 _connect_mutex），让 extra stream 先 Consume() 到 64B 数据后再放开 SetConnected()；修复后 SetConnected() 会基于 _atomic_local_consumed 立刻补发首个 FEEDBACK（consumed_size=64），从而使服务端 extra stream的第二次 StreamWrite(1B) 不再长期 EAGAIN，避免错误反压。如果没有这个修复，这个单测会直接因为错误反压而hang住。
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@brpc.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@brpc.apache.org
For additional commands, e-mail: dev-help@brpc.apache.org
Re: [PR] Make batch create stream SendFeedback thread safe (brpc)
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #3215:
URL: https://github.com/apache/brpc/pull/3215#discussion_r2888724175
##########
test/brpc_streaming_rpc_unittest.cpp:
##########
@@ -20,13 +20,16 @@
// Date: 2015/10/22 16:28:44
#include <gtest/gtest.h>
+#include <atomic>
#include "brpc/server.h"
#include "brpc/controller.h"
#include "brpc/channel.h"
+#include "brpc/callback.h"
#include "brpc/socket.h"
#include "brpc/stream_impl.h"
#include "brpc/policy/streaming_rpc_protocol.h"
+#include "bthread/countdown_event.h"
Review Comment:
`bthread/countdown_event.h` is included but not used in this test file, which adds unnecessary dependencies and compile time. Please remove it (or add the intended usage if it was meant for synchronization).
```suggestion
```
##########
test/brpc_streaming_rpc_unittest.cpp:
##########
@@ -78,6 +81,158 @@ class StreamingRpcTest : public testing::Test {
test::EchoResponse response;
};
+struct BatchStreamFeedbackRaceState {
+ brpc::StreamId server_first_stream_id{brpc::INVALID_STREAM_ID};
+ brpc::StreamId server_extra_stream_id{brpc::INVALID_STREAM_ID};
+ brpc::StreamId client_extra_stream_id{brpc::INVALID_STREAM_ID};
+
+ std::atomic<int> server_first_write_rc{-1};
+ std::atomic<int> server_second_write_rc{-1};
+ std::atomic<bool> client_got_first_msg{false};
+ std::atomic<bool> client_got_second_msg{false};
+ std::atomic<bool> server_write_done{false};
+ std::atomic<bool> rpc_done{false};
+
+ bthread_t server_send_tid{0};
+ std::atomic<bool> server_send_started{false};
+};
+
+class BatchStreamClientHandler : public brpc::StreamInputHandler {
+public:
+ explicit BatchStreamClientHandler(BatchStreamFeedbackRaceState* state)
+ : _state(state) {}
+
+ int on_received_messages(brpc::StreamId id,
+ butil::IOBuf* const messages[],
+ size_t size) override {
+ if (id != _state->client_extra_stream_id) {
+ // This test only cares about extra stream in batch creation.
+ return 0;
+ }
+ for (size_t i = 0; i < size; ++i) {
+ const size_t len = messages[i]->length();
+ messages[i]->clear();
+ // First payload: 64 bytes. Second payload: 1 byte.
+ if (len == 64) {
+ _state->client_got_first_msg.store(true, std::memory_order_release);
+ } else if (len == 1) {
+ _state->client_got_second_msg.store(true, std::memory_order_release);
+ }
+ }
+ return 0;
+ }
+
+ void on_idle_timeout(brpc::StreamId /*id*/) override {}
+
+ void on_closed(brpc::StreamId /*id*/) override {}
+
+ void on_failed(brpc::StreamId /*id*/, int /*error_code*/, const std::string& /*error_text*/) override {}
+
+private:
+ BatchStreamFeedbackRaceState* _state;
+};
+
+static void* SendTwoMessagesOnServerExtraStream(void* arg) {
+ auto* state = static_cast<BatchStreamFeedbackRaceState*>(arg);
+ const brpc::StreamId sid = state->server_extra_stream_id;
+
+ // Wait until server-side stream is connected.
+ const int64_t connect_deadline_us = butil::gettimeofday_us() + 2 * 1000 * 1000L;
+ bool connected = false;
+ while (butil::gettimeofday_us() < connect_deadline_us) {
+ brpc::SocketUniquePtr ptr;
+ if (brpc::Socket::Address(sid, &ptr) == 0) {
+ brpc::Stream* s = static_cast<brpc::Stream*>(ptr->conn());
+ if (s->_host_socket != NULL && s->_connected) {
+ connected = true;
+ break;
+ }
+ }
+ usleep(1000);
+ }
+
+ if (!connected) {
+ state->server_first_write_rc.store(ETIMEDOUT, std::memory_order_relaxed);
+ state->server_second_write_rc.store(ETIMEDOUT, std::memory_order_relaxed);
+ state->server_write_done.store(true, std::memory_order_release);
+ return NULL;
+ }
+
+ // 1) Send a payload exactly equal to max_buf_size(64).
+ {
+ std::string payload(64, 'a');
+ butil::IOBuf out;
+ out.append(payload);
+ state->server_first_write_rc.store(brpc::StreamWrite(sid, out), std::memory_order_relaxed);
+ }
+
+ // 2) Then send another byte. This write should become writable only after
+ // client sends FEEDBACK with consumed_size >= 64.
+ const int64_t write_deadline_us = butil::gettimeofday_us() + 2 * 1000 * 1000L;
+ int rc = -1;
+ while (butil::gettimeofday_us() < write_deadline_us) {
+ butil::IOBuf out;
+ out.append("b", 1);
+ rc = brpc::StreamWrite(sid, out);
+ if (rc == 0) {
+ break;
+ }
+ if (rc != EAGAIN) {
+ break;
+ }
+ const timespec duetime = butil::milliseconds_from_now(100);
+ (void)brpc::StreamWait(sid, &duetime);
+ }
+ state->server_second_write_rc.store(rc, std::memory_order_relaxed);
+ state->server_write_done.store(true, std::memory_order_release);
+ return NULL;
+}
+
+class MyServiceWithBatchStream : public test::EchoService {
+public:
+ MyServiceWithBatchStream(const brpc::StreamOptions& options,
+ BatchStreamFeedbackRaceState* state)
+ : _options(options), _state(state) {}
+
+ void Echo(::google::protobuf::RpcController* controller,
+ const ::test::EchoRequest* request,
+ ::test::EchoResponse* response,
+ ::google::protobuf::Closure* done) override {
+ brpc::ClosureGuard done_gurad(done);
+ response->set_message(request->message());
Review Comment:
Typo in variable name: `done_gurad` should be `done_guard` for clarity and consistency with `brpc::ClosureGuard`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@brpc.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@brpc.apache.org
For additional commands, e-mail: dev-help@brpc.apache.org
Re: [PR] Make batch create stream SendFeedback thread safe (brpc)
Posted by "wwbmmm (via GitHub)" <gi...@apache.org>.
wwbmmm commented on PR #3215:
URL: https://github.com/apache/brpc/pull/3215#issuecomment-4045308972
LGTM
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@brpc.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@brpc.apache.org
For additional commands, e-mail: dev-help@brpc.apache.org
Re: [PR] Make batch create stream SendFeedback thread safe (brpc)
Posted by "jenrryyou (via GitHub)" <gi...@apache.org>.
jenrryyou commented on PR #3215:
URL: https://github.com/apache/brpc/pull/3215#issuecomment-4131787282
@wwbmmm If there are no other issues, could you please help merge it?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@brpc.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@brpc.apache.org
For additional commands, e-mail: dev-help@brpc.apache.org