Copilot commented on code in PR #3650:
URL: https://github.com/apache/brpc/pull/3650#discussion_r4361984359
##########
src/brpc/selective_channel.cpp:
##########
@@ -286,11 +287,20 @@ inline int ChannelBalancer::SelectChannel(const
LoadBalancer::SelectIn& in,
}
int ChannelBalancer::CheckHealth() {
- BAIDU_SCOPED_LOCK(_mutex);
- for (ChannelToIdMap::const_iterator it = _chan_map.begin();
- it != _chan_map.end(); ++it) {
- if (!it->second->Failed() &&
- it->first->CheckHealth() == 0) {
+ std::vector<SocketId> ids;
+ {
+ BAIDU_SCOPED_LOCK(_mutex);
+ ids.reserve(_chan_map.size());
+ for (ChannelToIdMap::const_iterator it = _chan_map.begin();
+ it != _chan_map.end(); ++it) {
+ ids.push_back(it->second->id());
+ }
+ }
+ for (size_t i = 0; i < ids.size(); ++i) {
+ SocketUniquePtr ptr;
+ if (Socket::AddressFailedAsWell(ids[i], &ptr) >= 0 &&
+ !ptr->Failed() &&
+ static_cast<SubChannel*>(ptr->user())->chan->CheckHealth() == 0) {
Review Comment:
`static_cast<SubChannel*>(ptr->user())->chan` dereferences `ptr->user()`
without validating it is non-null and of the expected type. With the new
unlocked resolution approach, a resolved socket may be in a state where
`user()` is null/cleared (or otherwise not a `SubChannel`), which would crash
here. Add a defensive check for `ptr->user()` (and ideally validate it’s a
`SubChannel` instance) before dereferencing, skipping entries that don’t match.
##########
src/brpc/selective_channel.cpp:
##########
@@ -299,17 +309,31 @@ int ChannelBalancer::CheckHealth() {
void ChannelBalancer::Describe(std::ostream& os,
const DescribeOptions& options) {
- BAIDU_SCOPED_LOCK(_mutex);
if (!options.verbose) {
+ BAIDU_SCOPED_LOCK(_mutex);
os << _chan_map.size();
return;
}
- for (ChannelToIdMap::const_iterator it = _chan_map.begin();
- it != _chan_map.end(); ++it) {
- if (it != _chan_map.begin()) {
+ std::vector<SocketId> ids;
+ {
+ BAIDU_SCOPED_LOCK(_mutex);
+ ids.reserve(_chan_map.size());
+ for (ChannelToIdMap::const_iterator it = _chan_map.begin();
+ it != _chan_map.end(); ++it) {
+ ids.push_back(it->second->id());
+ }
+ }
+ bool first = true;
+ for (size_t i = 0; i < ids.size(); ++i) {
+ SocketUniquePtr ptr;
+ if (Socket::AddressFailedAsWell(ids[i], &ptr) < 0) {
+ continue;
+ }
+ if (!first) {
os << ' ';
}
- it->first->Describe(os, options);
+ static_cast<SubChannel*>(ptr->user())->chan->Describe(os, options);
Review Comment:
Same issue as in `CheckHealth()`: `ptr->user()` is dereferenced without
null/type validation. If `user()` is unset or not a `SubChannel`, this will
crash during verbose `Describe()`. Add a check to ensure `ptr->user()` is
present and expected before calling through to `chan->Describe(...)`, and skip
invalid entries.
##########
src/brpc/selective_channel.cpp:
##########
@@ -299,17 +309,31 @@ int ChannelBalancer::CheckHealth() {
void ChannelBalancer::Describe(std::ostream& os,
const DescribeOptions& options) {
- BAIDU_SCOPED_LOCK(_mutex);
if (!options.verbose) {
+ BAIDU_SCOPED_LOCK(_mutex);
os << _chan_map.size();
return;
}
- for (ChannelToIdMap::const_iterator it = _chan_map.begin();
- it != _chan_map.end(); ++it) {
- if (it != _chan_map.begin()) {
+ std::vector<SocketId> ids;
+ {
+ BAIDU_SCOPED_LOCK(_mutex);
+ ids.reserve(_chan_map.size());
+ for (ChannelToIdMap::const_iterator it = _chan_map.begin();
+ it != _chan_map.end(); ++it) {
+ ids.push_back(it->second->id());
+ }
+ }
+ bool first = true;
+ for (size_t i = 0; i < ids.size(); ++i) {
+ SocketUniquePtr ptr;
+ if (Socket::AddressFailedAsWell(ids[i], &ptr) < 0) {
+ continue;
+ }
Review Comment:
Same issue as in `CheckHealth()`: `ptr->user()` is dereferenced without
null/type validation. If `user()` is unset or not a `SubChannel`, this will
crash during verbose `Describe()`. Add a check to ensure `ptr->user()` is
present and expected before calling through to `chan->Describe(...)`, and skip
invalid entries.
##########
test/brpc_channel_unittest.cpp:
##########
@@ -4086,4 +4146,43 @@ TEST_F(ChannelTest,
selective_channel_add_failure_ownership) {
ASSERT_EQ(0, CountingSubChannel::s_alive);
}
+TEST(SelectiveChannelTest, describe_does_not_block_add_channel) {
+ brpc::SelectiveChannel channel;
+ ASSERT_EQ(0, channel.Init("rr", nullptr));
+
+ BlockingDescribeChannel* blocking = new BlockingDescribeChannel;
+ brpc::SelectiveChannel::ChannelHandle first_handle;
+ ASSERT_EQ(0, channel.AddChannel(blocking, &first_handle));
+
+ SelectiveDescribeArgs describe_args;
+ describe_args.channel = &channel;
+ pthread_t describe_tid;
+ ASSERT_EQ(0, pthread_create(&describe_tid, nullptr,
+ RunSelectiveDescribe, &describe_args));
+ for (int i = 0; i < 1000 && !blocking->started(); ++i) {
+ bthread_usleep(1000);
+ }
+ ASSERT_TRUE(blocking->started());
+
+ SelectiveAddArgs add_args;
+ add_args.channel = &channel;
+ pthread_t add_tid;
+ ASSERT_EQ(0, pthread_create(&add_tid, nullptr,
+ RunSelectiveAdd, &add_args));
+ for (int i = 0; i < 100 && !add_args.done.load(
+ butil::memory_order_acquire); ++i) {
+ bthread_usleep(1000);
+ }
+ const bool add_finished = add_args.done.load(butil::memory_order_acquire);
+
+ blocking->Release();
+ ASSERT_EQ(0, pthread_join(describe_tid, nullptr));
+ ASSERT_EQ(0, pthread_join(add_tid, nullptr));
+ EXPECT_TRUE(add_finished);
+ ASSERT_EQ(0, add_args.rc);
+
+ channel.RemoveAndDestroyChannel(add_args.handle);
+ channel.RemoveAndDestroyChannel(first_handle);
Review Comment:
`ASSERT_TRUE(blocking->started())` can cause the test process to hang: if
this assertion fails, `blocking->Release()` is never called and the
`Describe()` thread can spin forever. Restructure to guarantee cleanup (always
call `Release()` and `pthread_join()` via a scope guard / RAII cleanup), and
prefer `EXPECT_TRUE` here so teardown still runs.
##########
test/brpc_channel_unittest.cpp:
##########
@@ -4086,4 +4146,43 @@ TEST_F(ChannelTest,
selective_channel_add_failure_ownership) {
ASSERT_EQ(0, CountingSubChannel::s_alive);
}
+TEST(SelectiveChannelTest, describe_does_not_block_add_channel) {
+ brpc::SelectiveChannel channel;
+ ASSERT_EQ(0, channel.Init("rr", nullptr));
+
+ BlockingDescribeChannel* blocking = new BlockingDescribeChannel;
+ brpc::SelectiveChannel::ChannelHandle first_handle;
+ ASSERT_EQ(0, channel.AddChannel(blocking, &first_handle));
+
+ SelectiveDescribeArgs describe_args;
+ describe_args.channel = &channel;
+ pthread_t describe_tid;
+ ASSERT_EQ(0, pthread_create(&describe_tid, nullptr,
+ RunSelectiveDescribe, &describe_args));
+ for (int i = 0; i < 1000 && !blocking->started(); ++i) {
+ bthread_usleep(1000);
+ }
+ ASSERT_TRUE(blocking->started());
+
+ SelectiveAddArgs add_args;
+ add_args.channel = &channel;
+ pthread_t add_tid;
+ ASSERT_EQ(0, pthread_create(&add_tid, nullptr,
+ RunSelectiveAdd, &add_args));
+ for (int i = 0; i < 100 && !add_args.done.load(
+ butil::memory_order_acquire); ++i) {
Review Comment:
The `AddChannel()` completion wait is capped at ~100ms, which is prone to CI
flakiness under load/scheduling delays even if `AddChannel()` is not blocked by
`Describe()`. Increase the timeout substantially (e.g., >= 1s) and/or use a
more deterministic synchronization mechanism (condition variable/latch) so the
test asserts the intended non-blocking property without relying on tight
polling.
--
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]
