Merge branch 'apache:master' into master

zchuango · web-flow · commit cbdbb52c4269 · 2026-01-26T14:16:42.000+08:00
diff --git a/CMakeLists.txt b/CMakeLists.txt
@@ -40,7 +40,7 @@ if(POLICY CMP0042)
     cmake_policy(SET CMP0042 NEW)
 endif()
 
-set(BRPC_VERSION 1.15.0)
+set(BRPC_VERSION 1.16.0)
 
 SET(CPACK_GENERATOR "DEB")
 SET(CPACK_DEBIAN_PACKAGE_MAINTAINER "brpc authors")
diff --git a/MODULE.bazel b/MODULE.bazel
@@ -1,6 +1,6 @@
 module(
   name = 'brpc',
-  version = '1.15.0',
+  version = '1.16.0',
   compatibility_level = 1,
 )
 
diff --git a/NOTICE b/NOTICE
@@ -1,5 +1,5 @@
 Apache bRPC
-Copyright 2018-2025 The Apache Software Foundation
+Copyright 2018-2026 The Apache Software Foundation
 
 This product includes software developed at
 The Apache Software Foundation (http://www.apache.org/).
diff --git a/RELEASE_VERSION b/RELEASE_VERSION
@@ -1 +1 @@
-1.15.0
+1.16.0
diff --git a/package/rpm/brpc.spec b/package/rpm/brpc.spec
@@ -18,7 +18,7 @@
 #
 
 Name:		brpc
-Version:	1.15.0
+Version:	1.16.0
 Release:	1%{?dist}
 Summary:	Industrial-grade RPC framework using C++ Language.
 
diff --git a/src/brpc/channel.cpp b/src/brpc/channel.cpp
@@ -77,6 +77,8 @@ ChannelSSLOptions* ChannelOptions::mutable_ssl_options() {
 static ChannelSignature ComputeChannelSignature(const ChannelOptions& opt) {
     if (opt.auth == NULL &&
         !opt.has_ssl_options() &&
+        opt.client_host.empty() &&
+        opt.device_name.empty() &&
         opt.connection_group.empty() &&
         opt.hc_option.health_check_path.empty()) {
         // Returning zeroized result by default is more intuitive for users.
@@ -94,6 +96,14 @@ static ChannelSignature ComputeChannelSignature(const ChannelOptions& opt) {
             buf.append("|conng=");
             buf.append(opt.connection_group);
         }
+        if (!opt.client_host.empty()) {
+            buf.append("|clih=");
+            buf.append(opt.client_host);
+        }
+        if (!opt.device_name.empty()) {
+            buf.append("|devn=");
+            buf.append(opt.device_name);
+        }
         if (opt.auth) {
             buf.append("|auth=");
             buf.append((char*)&opt.auth, sizeof(opt.auth));
@@ -362,14 +372,27 @@ int Channel::InitSingle(const butil::EndPoint& server_addr_and_port,
         LOG(ERROR) << "Invalid port=" << port;
         return -1;
     }
+    butil::EndPoint client_endpoint;
+    if (!_options.client_host.empty() &&
+        butil::str2ip(_options.client_host.c_str(), &client_endpoint.ip) != 0 &&
+        butil::hostname2ip(_options.client_host.c_str(), &client_endpoint.ip) != 0) {
+        LOG(ERROR) << "Invalid client host=`" << _options.client_host << '\'';
+        return -1;
+    }
     _server_address = server_addr_and_port;
     const ChannelSignature sig = ComputeChannelSignature(_options);
     std::shared_ptr<SocketSSLContext> ssl_ctx;
     if (CreateSocketSSLContext(_options, &ssl_ctx) != 0) {
         return -1;
     }
+    SocketOptions opt;
+    opt.local_side = client_endpoint;
+    opt.initial_ssl_ctx = ssl_ctx;
+    opt.use_rdma = _options.use_rdma;
+    opt.hc_option = _options.hc_option;
+    opt.device_name = _options.device_name;
     if (SocketMapInsert(SocketMapKey(server_addr_and_port, sig),
-                        &_server_id, ssl_ctx, _options.use_rdma, _options.hc_option) != 0) {
+                        &_server_id, opt) != 0) {
         LOG(ERROR) << "Fail to insert into SocketMap";
         return -1;
     }
@@ -397,6 +420,13 @@ int Channel::Init(const char* ns_url,
             _options.mutable_ssl_options()->sni_name = _service_name;
         }
     }
+    butil::EndPoint client_endpoint;
+    if (!_options.client_host.empty() &&
+        butil::str2ip(_options.client_host.c_str(), &client_endpoint.ip) != 0 &&
+        butil::hostname2ip(_options.client_host.c_str(), &client_endpoint.ip) != 0) {
+        LOG(ERROR) << "Invalid client host=`" << _options.client_host << '\'';
+        return -1;
+    }
     std::unique_ptr<LoadBalancerWithNaming> lb(new (std::nothrow)
                                                    LoadBalancerWithNaming);
     if (NULL == lb) {
@@ -406,10 +436,13 @@ int Channel::Init(const char* ns_url,
     GetNamingServiceThreadOptions ns_opt;
     ns_opt.succeed_without_server = _options.succeed_without_server;
     ns_opt.log_succeed_without_server = _options.log_succeed_without_server;
-    ns_opt.use_rdma = _options.use_rdma;
+    ns_opt.socket_option.use_rdma = _options.use_rdma;
     ns_opt.channel_signature = ComputeChannelSignature(_options);
-    ns_opt.hc_option =  _options.hc_option;
-    if (CreateSocketSSLContext(_options, &ns_opt.ssl_ctx) != 0) {
+    ns_opt.socket_option.hc_option =  _options.hc_option;
+    ns_opt.socket_option.local_side = client_endpoint;
+    ns_opt.socket_option.device_name = _options.device_name;
+    if (CreateSocketSSLContext(_options,
+                               &ns_opt.socket_option.initial_ssl_ctx) != 0) {
         return -1;
     }
     if (lb->Init(ns_url, lb_name, _options.ns_filter, &ns_opt) != 0) {
diff --git a/src/brpc/channel.h b/src/brpc/channel.h
@@ -148,6 +148,16 @@ struct ChannelOptions {
     // Its priority is higher than FLAGS_health_check_path and FLAGS_health_check_timeout_ms.
     // When it is not set, FLAGS_health_check_path and FLAGS_health_check_timeout_ms will take effect.
     HealthCheckOption hc_option;
+
+    // IP address or host name of the client.
+    // if the client_host is "", the client IP address is determined by the OS.
+    // Default: ""
+    std::string client_host;
+
+    // The device name of the client's network adapter.
+    // if the device_name is "", the flow control is determined by the OS.
+    // Default: ""
+    std::string device_name;
 private:
     // SSLOptions is large and not often used, allocate it on heap to
     // prevent ChannelOptions from being bloated in most cases.
diff --git a/src/brpc/details/naming_service_thread.cpp b/src/brpc/details/naming_service_thread.cpp
@@ -125,8 +125,8 @@ void NamingServiceThread::Actions::ResetServers(
         //       Socket. SocketMapKey may be passed through AddWatcher. Make sure
         //       to pick those Sockets with the right settings during OnAddedServers
         const SocketMapKey key(_added[i], _owner->_options.channel_signature);
-        CHECK_EQ(0, SocketMapInsert(key, &tagged_id.id, _owner->_options.ssl_ctx,
-                                    _owner->_options.use_rdma, _owner->_options.hc_option));
+        CHECK_EQ(0, SocketMapInsert(key, &tagged_id.id, 
+                                    _owner->_options.socket_option));
         _added_sockets.push_back(tagged_id);
     }
 
diff --git a/src/brpc/details/naming_service_thread.h b/src/brpc/details/naming_service_thread.h
@@ -44,15 +44,14 @@ class NamingServiceWatcher {
 struct GetNamingServiceThreadOptions {
     GetNamingServiceThreadOptions()
         : succeed_without_server(false)
-        , log_succeed_without_server(true)
-        , use_rdma(false) {}
+        , log_succeed_without_server(true) {
+    socket_option.use_rdma = false;
+}
     
     bool succeed_without_server;
     bool log_succeed_without_server;
-    bool use_rdma;
-    HealthCheckOption hc_option;
     ChannelSignature channel_signature;
-    std::shared_ptr<SocketSSLContext> ssl_ctx;
+    SocketOptions socket_option;
 };
 
 // A dedicated thread to map a name to ServerIds
diff --git a/src/brpc/selective_channel.cpp b/src/brpc/selective_channel.cpp
@@ -41,10 +41,13 @@ typedef std::map<ChannelBase*, Socket*> ChannelToIdMap;
 class SubChannel : public SocketUser {
 public:
     ChannelBase* chan;
+    ChannelOwnership ownership;
 
     // internal channel is deleted after the fake Socket is SetFailed
     void BeforeRecycle(Socket*) {
-        delete chan;
+        if (ownership == OWNS_CHANNEL) {
+            delete chan;
+        }
         delete this;
     }
 
@@ -83,7 +86,8 @@ class ChannelBalancer : public SharedLoadBalancer {
     ChannelBalancer() {}
     ~ChannelBalancer();
     int Init(const char* lb_name);
-    int AddChannel(ChannelBase* sub_channel, const std::string& tag,
+    int AddChannel(ChannelBase* sub_channel,
+                   const SelectiveChannel::SubChannelOptions& subopt,
                    SelectiveChannel::ChannelHandle* handle);
     void RemoveAndDestroyChannel(const SelectiveChannel::ChannelHandle& handle);
     int SelectChannel(const LoadBalancer::SelectIn& in, SelectOut* out);
@@ -168,7 +172,8 @@ int ChannelBalancer::Init(const char* lb_name) {
     return SharedLoadBalancer::Init(lb_name);
 }
 
-int ChannelBalancer::AddChannel(ChannelBase* sub_channel, const std::string& tag,
+int ChannelBalancer::AddChannel(ChannelBase* sub_channel,
+                                const SelectiveChannel::SubChannelOptions& subopt,
                                 SelectiveChannel::ChannelHandle* handle) {
     if (NULL == sub_channel) {
         LOG(ERROR) << "Parameter[sub_channel] is NULL";
@@ -185,6 +190,7 @@ int ChannelBalancer::AddChannel(ChannelBase* sub_channel, const std::string& tag
         return -1;
     }
     sub_chan->chan = sub_channel;
+    sub_chan->ownership = subopt.ownership;
     SocketId sock_id;
     SocketOptions options;
     options.user = sub_chan;
@@ -206,7 +212,7 @@ int ChannelBalancer::AddChannel(ChannelBase* sub_channel, const std::string& tag
                    << sock_id << " is disabled";
         return -1;
     }
-    if (!AddServer(ServerId(sock_id, tag))) {
+    if (!AddServer(ServerId(sock_id, subopt.tag))) {
         LOG(ERROR) << "Duplicated sub_channel=" << sub_channel;
         // sub_chan will be deleted when the socket is recycled.
         ptr->SetFailed();
@@ -215,10 +221,10 @@ int ChannelBalancer::AddChannel(ChannelBase* sub_channel, const std::string& tag
         return -1;
     }
     // The health-check-related reference has been held on created.
-    _chan_map[sub_channel]= ptr.get();
+    _chan_map[sub_channel] = ptr.get();
     if (handle) {
         handle->id = sock_id;
-        handle->tag = tag;
+        handle->tag = subopt.tag;
     }
     return 0;
 }
@@ -532,20 +538,15 @@ bool SelectiveChannel::initialized() const {
 }
 
 int SelectiveChannel::AddChannel(ChannelBase* sub_channel,
-                                 ChannelHandle* handle) {
-    return AddChannel(sub_channel, "", handle);
-}
-
-int SelectiveChannel::AddChannel(ChannelBase* sub_channel,
-                                 const std::string& tag,
+                                 const SubChannelOptions& option,
                                  ChannelHandle* handle) {
     schan::ChannelBalancer* lb =
         static_cast<schan::ChannelBalancer*>(_chan._lb.get());
     if (lb == NULL) {
         LOG(ERROR) << "You must call Init() to initialize a SelectiveChannel";
         return -1;
     }
-    return lb->AddChannel(sub_channel, tag, handle);
+    return lb->AddChannel(sub_channel, option, handle);
 }
 
 void SelectiveChannel::RemoveAndDestroyChannel(const ChannelHandle& handle) {
diff --git a/src/brpc/selective_channel.h b/src/brpc/selective_channel.h
@@ -56,6 +56,11 @@ class SelectiveChannel : public ChannelBase/*non-copyable*/ {
         std::string tag;
     };
 
+    struct SubChannelOptions {
+        std::string tag;
+        ChannelOwnership ownership = OWNS_CHANNEL;
+    };
+
     SelectiveChannel();
     ~SelectiveChannel();
 
@@ -69,8 +74,16 @@ class SelectiveChannel : public ChannelBase/*non-copyable*/ {
     // On success, handle is set with the key for removal.
     // NOTE: Different from pchan, schan can add channels at any time.
     // Returns 0 on success, -1 otherwise.
-    int AddChannel(ChannelBase* sub_channel, ChannelHandle* handle);
-    int AddChannel(ChannelBase* sub_channel, const std::string& tag, ChannelHandle* handle);
+    int AddChannel(ChannelBase* sub_channel, ChannelHandle* handle) {
+        return AddChannel(sub_channel, SubChannelOptions(), handle);
+    }
+    int AddChannel(ChannelBase* sub_channel, const std::string& tag, ChannelHandle* handle) {
+        SubChannelOptions option;
+        option.tag = tag;
+        return AddChannel(sub_channel, option, handle);
+    }
+    int AddChannel(ChannelBase* sub_channel, const SubChannelOptions& option,
+                   ChannelHandle* handle);
 
     // Remove and destroy the sub_channel associated with `handle'.
     void RemoveAndDestroyChannel(const ChannelHandle& handle);
diff --git a/src/brpc/socket.cpp b/src/brpc/socket.cpp
@@ -728,7 +728,8 @@ int Socket::OnCreated(const SocketOptions& options) {
     _keytable_pool = options.keytable_pool;
     _tos = 0;
     _remote_side = options.remote_side;
-    _local_side = butil::EndPoint();
+    _local_side = options.local_side;
+    _device_name = options.device_name;
     _on_edge_triggered_events = options.on_edge_triggered_events;
     _user = options.user;
     _conn = options.conn;
@@ -1296,7 +1297,25 @@ int Socket::Connect(const timespec* abstime,
     CHECK_EQ(0, butil::make_close_on_exec(sockfd));
     // We need to do async connect (to manage the timeout by ourselves).
     CHECK_EQ(0, butil::make_non_blocking(sockfd));
-    
+    if (!_device_name.empty()) {
+        if (setsockopt(sockfd, SOL_SOCKET, SO_BINDTODEVICE,
+                       _device_name.c_str(), _device_name.size()) < 0) {
+            PLOG(ERROR) << "Fail to set SO_BINDTODEVICE of fd=" << sockfd
+                        << " to device_name=" << _device_name;
+            return -1;
+        }
+    }
+    if (local_side().ip != butil::IP_ANY) {
+        struct sockaddr_storage cli_addr;
+        if (butil::endpoint2sockaddr(local_side(), &cli_addr, &addr_size) != 0) {
+            PLOG(ERROR) << "Fail to get client sockaddr";
+            return -1;
+        }
+        if (::bind(sockfd, (struct sockaddr*)&cli_addr, addr_size) != 0) {
+            PLOG(ERROR) << "Fail to bind client socket, errno=" << strerror(errno);
+            return -1;
+        }
+    }
     const int rc = ::connect(
         sockfd, (struct sockaddr*)&serv_addr, addr_size);
     if (rc != 0 && errno != EINPROGRESS) {
@@ -2811,6 +2830,7 @@ int Socket::GetPooledSocket(SocketUniquePtr* pooled_socket) {
     if (socket_pool == NULL) {
         SocketOptions opt;
         opt.remote_side = remote_side();
+        opt.local_side = butil::EndPoint(local_side().ip, 0);
         opt.user = user();
         opt.on_edge_triggered_events = _on_edge_triggered_events;
         opt.initial_ssl_ctx = _ssl_ctx;
@@ -2912,6 +2932,7 @@ int Socket::GetShortSocket(SocketUniquePtr* short_socket) {
     SocketId id;
     SocketOptions opt;
     opt.remote_side = remote_side();
+    opt.local_side = butil::EndPoint(local_side().ip, 0);
     opt.user = user();
     opt.on_edge_triggered_events = _on_edge_triggered_events;
     opt.initial_ssl_ctx = _ssl_ctx;
diff --git a/src/brpc/socket.h b/src/brpc/socket.h
@@ -250,6 +250,8 @@ struct SocketOptions {
     // user->BeforeRecycle() before recycling.
     int fd{-1};
     butil::EndPoint remote_side;
+    butil::EndPoint local_side;
+    std::string device_name;
     // If `connect_on_create' is true and `fd' is less than 0,
     // a client connection will be established to remote_side()
     // regarding deadline `connect_abstime' when Socket is being created.
@@ -830,6 +832,9 @@ friend void DereferenceSocket(Socket*);
     // Address of self. Initialized in ResetFileDescriptor().
     butil::EndPoint _local_side;
 
+    // The device name of the client's network adapter.
+    std::string _device_name;
+
     // Called when edge-triggered events happened on `_fd'. Read comments
     // of EventDispatcher::AddConsumer (event_dispatcher.h)
     // carefully before implementing the callback.
diff --git a/src/brpc/socket_map.cpp b/src/brpc/socket_map.cpp
@@ -90,11 +90,9 @@ SocketMap* get_or_new_client_side_socket_map() {
 }
 
 int SocketMapInsert(const SocketMapKey& key, SocketId* id,
-                    const std::shared_ptr<SocketSSLContext>& ssl_ctx,
-                    bool use_rdma,
-                    const HealthCheckOption& hc_option) {
-    return get_or_new_client_side_socket_map()->Insert(key, id, ssl_ctx, use_rdma, hc_option);
-}    
+                    SocketOptions& opt) {
+    return get_or_new_client_side_socket_map()->Insert(key, id, opt);
+}
 
 int SocketMapFind(const SocketMapKey& key, SocketId* id) {
     SocketMap* m = get_client_side_socket_map();
@@ -227,9 +225,7 @@ void SocketMap::ShowSocketMapInBvarIfNeed() {
 }
 
 int SocketMap::Insert(const SocketMapKey& key, SocketId* id,
-                      const std::shared_ptr<SocketSSLContext>& ssl_ctx,
-                      bool use_rdma,
-                      const HealthCheckOption& hc_option) {
+                      SocketOptions& opt) {
     ShowSocketMapInBvarIfNeed();
 
     std::unique_lock<butil::Mutex> mu(_mutex);
@@ -249,11 +245,7 @@ int SocketMap::Insert(const SocketMapKey& key, SocketId* id,
         sc = NULL;
     }
     SocketId tmp_id;
-    SocketOptions opt;
     opt.remote_side = key.peer.addr;
-    opt.initial_ssl_ctx = ssl_ctx;
-    opt.use_rdma = use_rdma;
-    opt.hc_option = hc_option;
     if (_options.socket_creator->CreateSocket(opt, &tmp_id) != 0) {
         PLOG(FATAL) << "Fail to create socket to " << key.peer;
         return -1;
diff --git a/src/brpc/socket_map.h b/src/brpc/socket_map.h
diff --git a/src/bthread/bthread.h b/src/bthread/bthread.h
diff --git a/src/bthread/condition_variable.h b/src/bthread/condition_variable.h
diff --git a/test/brpc_server_unittest.cpp b/test/brpc_server_unittest.cpp
diff --git a/test/bthread_cond_unittest.cpp b/test/bthread_cond_unittest.cpp

Original file line number	Diff line number	Diff line change
`@@ -1,6 +1,6 @@`
`1`	`1`	`module(`
`2`	`2`	`name = 'brpc',`
`3`		`- version = '1.15.0',`
	`3`	`+ version = '1.16.0',`
`4`	`4`	`compatibility_level = 1,`
`5`	`5`	`)`
`6`	`6`
Original file line number	Diff line number	Diff line change
`@@ -18,7 +18,7 @@`
`18`	`18`	`#`
`19`	`19`
`20`	`20`	`Name: brpc`
`21`		`-Version: 1.15.0`
	`21`	`+Version: 1.16.0`
`22`	`22`	`Release: 1%{?dist}`
`23`	`23`	`Summary: Industrial-grade RPC framework using C++ Language.`
`24`	`24`