Copilot commented on code in PR #3493:
URL: https://github.com/apache/brpc/pull/3493#discussion_r3864706493
##########
src/brpc/stream.cpp:
##########
@@ -1034,6 +1039,15 @@ int StreamAccept(StreamIds& response_streams,
Controller& cntl,
LOG(ERROR) << "No stream along with this request";
return -1;
}
+ const int64_t stream_count = static_cast<int64_t>(
+ cntl._remote_stream_settings->extra_stream_ids_size()) + 1;
+ if (stream_count > FLAGS_stream_max_streams_per_request) {
+ LOG(ERROR) << "Reject " << stream_count
+ << " streams in one request, exceeding "
+ "-stream_max_streams_per_request="
+ << FLAGS_stream_max_streams_per_request;
+ return -1;
+ }
Review Comment:
When the stream count exceeds the limit, StreamAccept returns -1 but does
not set the Controller failure. The client then only sees the generic "The
server didn't accept the stream" error, which makes debugging harder and hides
the configured limit; set cntl.SetFailed(EREQUEST, ...) before returning so the
rejection reason is propagated to the RPC caller.
##########
test/brpc_streaming_rpc_unittest.cpp:
##########
@@ -1249,6 +1249,74 @@ class MyServiceWithMismatchedExtraStreamIds : public
test::EchoService {
int _adjustment;
};
+class MyServiceWithStreamCountLimit : public test::EchoService {
+public:
+ void Echo(::google::protobuf::RpcController* controller,
+ const ::test::EchoRequest* request,
+ ::test::EchoResponse* response,
+ ::google::protobuf::Closure* done) override {
+ brpc::ClosureGuard done_guard(done);
+ brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);
+ response->set_message(request->message());
+
+ brpc::StreamIds response_streams;
+ accept_result = brpc::StreamAccept(response_streams, *cntl, nullptr);
+ accepted_streams = response_streams.size();
+ }
+
+ int accept_result{0};
+ size_t accepted_streams{0};
+};
+
+TEST_F(StreamingRpcTest, limit_streams_accepted_per_request) {
+ std::string old_stream_limit;
+ ASSERT_TRUE(GFLAGS_NAMESPACE::GetCommandLineOption(
+ "stream_max_streams_per_request", &old_stream_limit));
+ BRPC_SCOPE_EXIT {
+ GFLAGS_NAMESPACE::SetCommandLineOption(
+ "stream_max_streams_per_request", old_stream_limit.c_str());
+ };
+ ASSERT_FALSE(GFLAGS_NAMESPACE::SetCommandLineOption(
+ "stream_max_streams_per_request", "2").empty());
+
+ brpc::Server server;
+ MyServiceWithStreamCountLimit service;
+ ASSERT_EQ(0, server.AddService(
+ &service, brpc::SERVER_DOESNT_OWN_SERVICE));
+ ASSERT_EQ(0, server.Start(0, nullptr));
+
+ brpc::Channel channel;
+ ASSERT_EQ(0, channel.Init(server.listen_address(), nullptr));
+
+ for (size_t stream_count : {2u, 3u, 2u}) {
+ brpc::Controller cntl;
+ brpc::StreamIds request_streams;
+ ASSERT_EQ(0, brpc::StreamCreate(
+ request_streams, stream_count, cntl, nullptr));
+ ASSERT_EQ(stream_count, request_streams.size());
+
+ test::EchoService_Stub stub(&channel);
+ stub.Echo(&cntl, &request, &response, nullptr);
+ if (stream_count == 2) {
+ ASSERT_FALSE(cntl.Failed()) << cntl.ErrorText();
+ ASSERT_EQ(0, service.accept_result);
+ ASSERT_EQ(stream_count, service.accepted_streams);
+ } else {
+ ASSERT_TRUE(cntl.Failed());
+ ASSERT_EQ(brpc::EREQUEST, cntl.ErrorCode());
+ ASSERT_EQ(-1, service.accept_result);
+ ASSERT_EQ(0u, service.accepted_streams);
+ }
Review Comment:
After making accept_result/accepted_streams atomic, the assertions should
read them via load() (and ideally use a consistent memory_order) instead of
accessing the atomic objects directly.
##########
test/brpc_streaming_rpc_unittest.cpp:
##########
@@ -1249,6 +1249,74 @@ class MyServiceWithMismatchedExtraStreamIds : public
test::EchoService {
int _adjustment;
};
+class MyServiceWithStreamCountLimit : public test::EchoService {
+public:
+ void Echo(::google::protobuf::RpcController* controller,
+ const ::test::EchoRequest* request,
+ ::test::EchoResponse* response,
+ ::google::protobuf::Closure* done) override {
+ brpc::ClosureGuard done_guard(done);
+ brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);
+ response->set_message(request->message());
+
+ brpc::StreamIds response_streams;
+ accept_result = brpc::StreamAccept(response_streams, *cntl, nullptr);
+ accepted_streams = response_streams.size();
+ }
+
+ int accept_result{0};
+ size_t accepted_streams{0};
Review Comment:
MyServiceWithStreamCountLimit writes accept_result/accepted_streams in the
server thread and the test reads them from the client thread without any
explicit synchronization. Make these fields atomic (or otherwise synchronized)
to avoid undefined behavior/data races and potential TSAN flakes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
