This is an automated email from the ASF dual-hosted git repository.
chenBright 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 962743de Fix inconsistent ownership handling in
SelectiveChannel::AddChannel on failure (#3595)
962743de is described below
commit 962743de08bfaa5cfb98a9339eaea673dba601a5
Author: NeutralMilkHotel <[email protected]>
AuthorDate: Fri Oct 23 17:45:12 2026 +0800
Fix inconsistent ownership handling in SelectiveChannel::AddChannel on
failure (#3595)
* Fix inconsistent ownership handling in SelectiveChannel::AddChannel on
failure (#1408)
* Fix double-free of sub channel on SelectiveChannel fake socket creation
failure
* Ensure SocketUser is cleaned up when Socket creation fails at IOEvent init
* Fix double-free of SubPartitionChannel on SelectiveChannel::AddChannel
failure (#1408)
* Update user attachment logic on IOEvent init failure
Attach user before returning on IOEvent initialization failure to ensure
proper cleanup.
Co-authored-by: Copilot Autofix powered by AI
<[email protected]>
* Fix SelectiveChannel::AddChannel ownership handling on failure
* Fix sub channel leak when SelectiveChannel::AddChannel is called before
Init
* Use ScopeGuard to centralise early-failure cleanup in
ChannelBalancer::AddChannel
---------
Co-authored-by: Copilot Autofix powered by AI
<[email protected]>
---
src/brpc/partition_channel.cpp | 5 +-
src/brpc/selective_channel.cpp | 31 ++++++++++-
src/brpc/selective_channel.h | 8 +++
src/brpc/socket.h | 7 +++
test/brpc_channel_unittest.cpp | 120 +++++++++++++++++++++++++++++++++++++++++
5 files changed, 168 insertions(+), 3 deletions(-)
diff --git a/src/brpc/partition_channel.cpp b/src/brpc/partition_channel.cpp
index 80633af2..28a6dd1a 100644
--- a/src/brpc/partition_channel.cpp
+++ b/src/brpc/partition_channel.cpp
@@ -313,7 +313,10 @@ public:
if (_schan->AddChannel(pchan, &pchan->handle) != 0) {
LOG(ERROR) << "Fail to add SubPartitionChannel=#"
<< part.num_partition_kinds;
- delete pchan;
+ // pchan is added with the default OWNS_CHANNEL: on failure
+ // AddChannel deletes pchan (at the call site or via
+ // SubChannel::BeforeRecycle), so do not delete it here
+ // (that would double-free).
continue;
}
_part_chan_map[part.num_partition_kinds] = pchan;
diff --git a/src/brpc/selective_channel.cpp b/src/brpc/selective_channel.cpp
index d670a947..5d2fbe89 100644
--- a/src/brpc/selective_channel.cpp
+++ b/src/brpc/selective_channel.cpp
@@ -24,6 +24,7 @@
#include "brpc/details/controller_private_accessor.h" // RPCSender
#include "brpc/errno.pb.h" // ERESPONSE
#include "brpc/nonreflectable_message.h"
+#include "butil/memory/scope_guard.h"
#include "brpc/selective_channel.h"
#include "brpc/global.h"
@@ -188,19 +189,30 @@ int ChannelBalancer::AddChannel(ChannelBase* sub_channel,
}
BAIDU_SCOPED_LOCK(_mutex);
if (_chan_map.find(sub_channel) != _chan_map.end()) {
+ // Duplicate: no new registration is created. Ownership is decided by
+ // the first successful registration, so ignore this call's ownership,
+ // don't delete (still referenced by the live registration) and don't
+ // reassign (a failing call should have no side effect).
LOG(ERROR) << "Duplicated sub_channel=" << sub_channel;
return -1;
}
SubChannel* sub_chan = new SubChannel;
sub_chan->chan = sub_channel;
sub_chan->ownership = subopt.ownership;
+ // The socket is deliberately left user-detached during Create so that
+ // any early failure path never triggers BeforeRecycle. This scope guard
+ // centralises the matching cleanup for those paths. It is dismissed
+ // right after set_user below, where ownership fully transfers to the
+ // socket and subsequent failures rely on SetFailed-driven recycling.
+ auto cleanup_subchan = butil::MakeScopeGuard([&]() noexcept {
+ sub_chan->BeforeRecycle(nullptr);
+ });
+
SocketId sock_id;
SocketOptions options;
- options.user = sub_chan;
options.health_check_interval_s = FLAGS_channel_check_interval;
if (Socket::Create(options, &sock_id) != 0) {
- delete sub_chan;
LOG(ERROR) << "Fail to create fake socket for sub channel";
return -1;
}
@@ -210,9 +222,19 @@ int ChannelBalancer::AddChannel(ChannelBase* sub_channel,
LOG(ERROR) << "Fail to address SocketId=" << sock_id;
return -1;
}
+ // The socket is fully created and referenced by `ptr'. Attach sub_chan
+ // now; from here on, failures SetFailed() the socket and its recycling
+ // runs SubChannel::BeforeRecycle, deleting sub_chan (and chan iff
+ // OWNS_CHANNEL).
+ ptr->set_user(sub_chan);
+ cleanup_subchan.dismiss();
if (rc > 0 && !ptr->HCEnabled()) {
LOG(ERROR) << "Health check of SocketId="
<< sock_id << " is disabled";
+ // sub_chan will be deleted when the socket is recycled.
+ ptr->SetFailed();
+ // Cancel health checking.
+ ptr->ReleaseHCRelatedReference();
return -1;
}
if (!AddServer(ServerId(sock_id, subopt.tag))) {
@@ -573,6 +595,11 @@ int SelectiveChannel::AddChannel(ChannelBase* sub_channel,
static_cast<schan::ChannelBalancer*>(_chan._lb.get());
if (lb == nullptr) {
LOG(ERROR) << "You must call Init() to initialize a SelectiveChannel";
+ // The ownership contract applies to every failure: take over an
+ // OWNS_CHANNEL sub channel even though it was never registered.
+ if (option.ownership == OWNS_CHANNEL) {
+ delete sub_channel;
+ }
return -1;
}
return lb->AddChannel(sub_channel, option, handle);
diff --git a/src/brpc/selective_channel.h b/src/brpc/selective_channel.h
index a728ac5e..576b71fd 100644
--- a/src/brpc/selective_channel.h
+++ b/src/brpc/selective_channel.h
@@ -72,6 +72,14 @@ public:
// Add a sub channel, which will be deleted along with schan or explicitly
// by RemoveAndDestroyChannel.
// On success, handle is set with the key for removal.
+ //
+ // Ownership (single source of truth: the FIRST successful registration):
+ // - OWNS_CHANNEL: schan owns sub_channel; caller must not delete it.
+ // - DOESNT_OWN_CHANNEL: caller owns sub_channel and must delete it.
+ // On failure the same rule applies to the registration being created:
schan
+ // deletes sub_channel iff its ownership is OWNS_CHANNEL.
+ // A duplicate sub_channel returns -1 and changes nothing; it stays owned
as
+ // its first registration decided.
// NOTE: Different from pchan, schan can add channels at any time.
// Returns 0 on success, -1 otherwise.
int AddChannel(ChannelBase* sub_channel, ChannelHandle* handle) {
diff --git a/src/brpc/socket.h b/src/brpc/socket.h
index da701e9a..94f85880 100644
--- a/src/brpc/socket.h
+++ b/src/brpc/socket.h
@@ -493,6 +493,13 @@ public:
// `user' parameter passed to Create().
SocketUser* user() const { return _user; }
+ // Attach the SocketUser after creation. Only safe before the SocketId
+ // is published to other threads (e.g. right after a successful
+ // Create+Address). Callers creating a socket with a null user and
+ // attaching it here get a single cleanup rule: creation failures never
+ // run SocketUser::BeforeRecycle, so the call site disposes the user.
+ void set_user(SocketUser* user) { _user = user; }
+
// `conn' parameter passed to Create()
void set_conn(SocketConnection* conn) { _conn = conn; }
SocketConnection* conn() const { return _conn; }
diff --git a/test/brpc_channel_unittest.cpp b/test/brpc_channel_unittest.cpp
index 3663efa6..f6155cbd 100644
--- a/test/brpc_channel_unittest.cpp
+++ b/test/brpc_channel_unittest.cpp
@@ -3966,4 +3966,124 @@ TEST_F(RateLimitedBackupPolicyTest,
OnRPCEndDrivesRatioDownAndReAllows) {
ASSERT_TRUE(p->DoBackup(nullptr));
}
+// Issue #1408: SelectiveChannel::AddChannel must take over (or not take over)
+// the sub channel consistently w.r.t. option.ownership, even on failure, so
+// that callers don't leak or double-free.
+class CountingSubChannel : public brpc::Channel {
+public:
+ static int s_alive;
+ CountingSubChannel() { ++s_alive; }
+ ~CountingSubChannel() override { --s_alive; }
+};
+int CountingSubChannel::s_alive = 0;
+
+TEST_F(ChannelTest, selective_channel_add_failure_ownership) {
+ // nullptr sub channel is rejected without touching ownership.
+ {
+ brpc::SelectiveChannel channel;
+ ASSERT_EQ(0, channel.Init("rr", nullptr));
+ EXPECT_EQ(-1, channel.AddChannel(nullptr, nullptr));
+ }
+ ASSERT_EQ(0, CountingSubChannel::s_alive);
+
+ // AddChannel before Init() honors the contract as well: an OWNS_CHANNEL
+ // sub channel is consumed, a DOESNT_OWN_CHANNEL one is left to the caller.
+ {
+ brpc::SelectiveChannel channel; // no Init()
+ CountingSubChannel* u = new CountingSubChannel();
+ ASSERT_EQ(0, u->Init("127.0.0.1:1", nullptr));
+ ASSERT_EQ(1, CountingSubChannel::s_alive);
+ brpc::SelectiveChannel::SubChannelOptions opt;
+ opt.ownership = brpc::OWNS_CHANNEL;
+ EXPECT_EQ(-1, channel.AddChannel(u, opt, nullptr));
+ EXPECT_EQ(0, CountingSubChannel::s_alive);
+ }
+ ASSERT_EQ(0, CountingSubChannel::s_alive);
+ {
+ brpc::SelectiveChannel channel; // no Init()
+ CountingSubChannel* u = new CountingSubChannel();
+ ASSERT_EQ(0, u->Init("127.0.0.1:1", nullptr));
+ ASSERT_EQ(1, CountingSubChannel::s_alive);
+ brpc::SelectiveChannel::SubChannelOptions opt;
+ opt.ownership = brpc::DOESNT_OWN_CHANNEL;
+ EXPECT_EQ(-1, channel.AddChannel(u, opt, nullptr));
+ EXPECT_EQ(1, CountingSubChannel::s_alive);
+ delete u;
+ EXPECT_EQ(0, CountingSubChannel::s_alive);
+ }
+ ASSERT_EQ(0, CountingSubChannel::s_alive);
+
+ // Plan 1: a duplicate AddChannel never reassigns ownership; the FIRST
+ // successful registration is the single source of truth.
+ // First reg = OWNS, duplicate with DOESNT_OWN must NOT downgrade: schan
+ // still owns and deletes the channel on removal; caller must NOT delete.
+ {
+ brpc::SelectiveChannel channel;
+ ASSERT_EQ(0, channel.Init("rr", nullptr));
+ brpc::SelectiveChannel::ChannelHandle h;
+ CountingSubChannel* c = new CountingSubChannel();
+ ASSERT_EQ(0, c->Init("127.0.0.1:1", nullptr));
+ ASSERT_EQ(1, CountingSubChannel::s_alive);
+ {
+ brpc::SelectiveChannel::SubChannelOptions opt;
+ opt.ownership = brpc::OWNS_CHANNEL;
+ ASSERT_EQ(0, channel.AddChannel(c, opt, &h));
+ }
+ {
+ brpc::SelectiveChannel::SubChannelOptions opt;
+ opt.ownership = brpc::DOESNT_OWN_CHANNEL;
+ EXPECT_EQ(-1, channel.AddChannel(c, opt, nullptr));
+ }
+ // Ownership unchanged (still OWNS): schan deletes c.
+ channel.RemoveAndDestroyChannel(h);
+ EXPECT_EQ(0, CountingSubChannel::s_alive);
+ }
+ ASSERT_EQ(0, CountingSubChannel::s_alive);
+
+ // First reg = DOESNT_OWN, duplicate with OWNS must NOT upgrade: the caller
+ // still owns the channel and must delete it; schan does not.
+ {
+ brpc::SelectiveChannel channel;
+ ASSERT_EQ(0, channel.Init("rr", nullptr));
+ brpc::SelectiveChannel::ChannelHandle h;
+ CountingSubChannel* d = new CountingSubChannel();
+ ASSERT_EQ(0, d->Init("127.0.0.1:1", nullptr));
+ ASSERT_EQ(1, CountingSubChannel::s_alive);
+ {
+ brpc::SelectiveChannel::SubChannelOptions opt;
+ opt.ownership = brpc::DOESNT_OWN_CHANNEL;
+ ASSERT_EQ(0, channel.AddChannel(d, opt, &h));
+ }
+ {
+ brpc::SelectiveChannel::SubChannelOptions opt;
+ opt.ownership = brpc::OWNS_CHANNEL;
+ EXPECT_EQ(-1, channel.AddChannel(d, opt, nullptr));
+ }
+ channel.RemoveAndDestroyChannel(h);
+ // Ownership unchanged (still DOESNT_OWN): schan does not delete d.
+ EXPECT_EQ(1, CountingSubChannel::s_alive);
+ delete d;
+ EXPECT_EQ(0, CountingSubChannel::s_alive);
+ }
+ ASSERT_EQ(0, CountingSubChannel::s_alive);
+
+ // OWNS_CHANNEL on success: schan takes over and deletes on removal.
+ {
+ brpc::SelectiveChannel channel;
+ ASSERT_EQ(0, channel.Init("rr", nullptr));
+ brpc::SelectiveChannel::ChannelHandle h;
+ CountingSubChannel* e = new CountingSubChannel();
+ ASSERT_EQ(0, e->Init("127.0.0.1:1", nullptr));
+ ASSERT_EQ(1, CountingSubChannel::s_alive);
+ {
+ brpc::SelectiveChannel::SubChannelOptions opt;
+ opt.ownership = brpc::OWNS_CHANNEL;
+ ASSERT_EQ(0, channel.AddChannel(e, opt, &h));
+ }
+ channel.RemoveAndDestroyChannel(h);
+ EXPECT_EQ(0, CountingSubChannel::s_alive);
+ }
+ ASSERT_EQ(0, CountingSubChannel::s_alive);
+}
+
} //namespace
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
