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 8b2410d2 Fix client source IP binding loss after socket revival (#3544)
8b2410d2 is described below
commit 8b2410d20b8b32c5700537a0ed1cdcdaa6bc491b
Author: Bright Chen <[email protected]>
AuthorDate: Tue Sep 15 21:55:07 2026 +0800
Fix client source IP binding loss after socket revival (#3544)
---
src/brpc/socket.cpp | 50 ++++++++---
src/brpc/socket.h | 11 ++-
test/brpc_socket_unittest.cpp | 197 ++++++++++++++++++++++++++++++++++++++++--
3 files changed, 240 insertions(+), 18 deletions(-)
diff --git a/src/brpc/socket.cpp b/src/brpc/socket.cpp
index a0b49662..57cfd229 100644
--- a/src/brpc/socket.cpp
+++ b/src/brpc/socket.cpp
@@ -742,6 +742,7 @@ int Socket::OnCreated(const SocketOptions& options) {
_tos = 0;
_remote_side = options.remote_side;
_local_side = options.local_side;
+ _bind_local_side = options.local_side;
_device_name = options.device_name;
_on_edge_triggered_events = options.on_edge_triggered_events;
_need_on_edge_trigger = options.need_on_edge_trigger;
@@ -797,6 +798,25 @@ int Socket::OnCreated(const SocketOptions& options) {
_http_request_method = HTTP_METHOD_GET;
CHECK(nullptr == _write_head.load(butil::memory_order_relaxed));
_is_write_shutdown = false;
+ // EndPoint::port is a type tag for IPv6/UDS, not a network port.
+ // Normalize only IP ports and leave Unix-domain addresses intact.
+ if (!butil::is_endpoint_extended(_bind_local_side)) {
+ _bind_local_side.port = 0;
+ } else if (butil::get_endpoint_type(_bind_local_side) == AF_INET6) {
+ sockaddr_storage addr{};
+ socklen_t addr_size = 0;
+ if (butil::endpoint2sockaddr(_bind_local_side, &addr, &addr_size) !=
0) {
+ SetFailed(EINVAL, "Fail to get client binding sockaddr from %s",
+ butil::endpoint2str(_bind_local_side).c_str());
+ return -1;
+ }
+ reinterpret_cast<sockaddr_in6*>(&addr)->sin6_port = 0;
+ if (butil::sockaddr2endpoint(&addr, addr_size, &_bind_local_side) !=
0) {
+ SetFailed(ENOMEM, "Fail to create client binding endpoint from %s",
+ butil::endpoint2str(_bind_local_side).c_str());
+ return -1;
+ }
+ }
int fd = options.fd;
if (!ValidFileDescriptor(fd) && options.connect_on_create) {
// Connect on create.
@@ -1015,6 +1035,7 @@ int Socket::WaitAndReset(int32_t expected_nref) {
}
_transport->Reset(expected_nref);
+ // Clear the runtime endpoint, keeping the configured binding intact.
_local_side = butil::EndPoint();
if (_ssl_session) {
SSL_free(_ssl_session);
@@ -1265,8 +1286,8 @@ int Socket::Connect(const timespec* abstime,
_ssl_state = SSL_OFF;
}
struct sockaddr_storage serv_addr;
- socklen_t addr_size = 0;
- if (butil::endpoint2sockaddr(remote_side(), &serv_addr, &addr_size) != 0) {
+ socklen_t serv_addr_size = 0;
+ if (butil::endpoint2sockaddr(remote_side(), &serv_addr, &serv_addr_size)
!= 0) {
PLOG(ERROR) << "Fail to get sockaddr";
return -1;
}
@@ -1294,19 +1315,20 @@ int Socket::Connect(const timespec* abstime,
return -1;
#endif
}
- if (local_side().ip != butil::IP_ANY) {
- struct sockaddr_storage cli_addr;
- if (butil::endpoint2sockaddr(local_side(), &cli_addr, &addr_size) !=
0) {
+ // Use the configured endpoint, not the runtime endpoint from
getsockname().
+ if (butil::is_endpoint_extended(_bind_local_side) || _bind_local_side.ip
!= butil::IP_ANY) {
+ struct sockaddr_storage cli_addr{};
+ socklen_t cli_addr_size = 0;
+ if (butil::endpoint2sockaddr(_bind_local_side, &cli_addr,
&cli_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);
+ if (::bind(sockfd, (struct sockaddr*)&cli_addr, cli_addr_size) != 0) {
+ PLOG(ERROR) << "Fail to bind client socket";
return -1;
}
}
- const int rc = ::connect(
- sockfd, (struct sockaddr*)&serv_addr, addr_size);
+ const int rc = ::connect(sockfd, (struct sockaddr*)&serv_addr,
serv_addr_size);
if (rc != 0 && errno != EINPROGRESS) {
PLOG(WARNING) << "Fail to connect to " << remote_side();
return -1;
@@ -2797,7 +2819,10 @@ int Socket::GetPooledSocket(SocketUniquePtr*
pooled_socket) {
if (socket_pool == nullptr) {
SocketOptions opt;
opt.remote_side = remote_side();
- opt.local_side = butil::EndPoint(local_side().ip, 0);
+ // Propagate the configured client binding (source IP + device) to
+ // pooled sub-sockets so that they keep the same binding policy.
+ opt.local_side = _bind_local_side;
+ opt.device_name = _device_name;
opt.user = user();
opt.on_edge_triggered_events = _on_edge_triggered_events;
opt.need_on_edge_trigger = _need_on_edge_trigger;
@@ -2900,7 +2925,10 @@ int Socket::GetShortSocket(SocketUniquePtr*
short_socket) {
SocketId id;
SocketOptions opt;
opt.remote_side = remote_side();
- opt.local_side = butil::EndPoint(local_side().ip, 0);
+ // Propagate the configured client binding (source IP + device) to short
+ // sub-sockets so that they keep the same binding policy.
+ opt.local_side = _bind_local_side;
+ opt.device_name = _device_name;
opt.user = user();
opt.on_edge_triggered_events = _on_edge_triggered_events;
opt.need_on_edge_trigger = _need_on_edge_trigger;
diff --git a/src/brpc/socket.h b/src/brpc/socket.h
index 6f6f52fb..97765f63 100644
--- a/src/brpc/socket.h
+++ b/src/brpc/socket.h
@@ -263,6 +263,9 @@ struct SocketOptions {
// user->BeforeRecycle() before recycling.
int fd{-1};
butil::EndPoint remote_side;
+ // Client source address. For IPv4/IPv6, the port is ignored for binding
+ // and the OS allocates a source port. IPv4 IP_ANY disables explicit
binding.
+ // Unix-domain addresses are preserved, including their paths.
butil::EndPoint local_side;
std::string device_name;
// If `connect_on_create' is true and `fd' is less than 0,
@@ -881,9 +884,15 @@ private:
// Address of peer. Initialized by SocketOptions.remote_side.
butil::EndPoint _remote_side;
- // Address of self. Initialized in ResetFileDescriptor().
+ // Runtime local endpoint. Updated in ResetFileDescriptor() and cleared
+ // in WaitAndReset().
butil::EndPoint _local_side;
+ // Client binding address from SocketOptions.local_side, preserved across
+ // health-check/revive. IPv4/IPv6 network ports are normalized to 0;
+ // Unix-domain addresses and extended endpoint type tags are preserved.
+ butil::EndPoint _bind_local_side;
+
// The device name of the client's network adapter.
std::string _device_name;
diff --git a/test/brpc_socket_unittest.cpp b/test/brpc_socket_unittest.cpp
index 740a1728..30e4e763 100644
--- a/test/brpc_socket_unittest.cpp
+++ b/test/brpc_socket_unittest.cpp
@@ -29,6 +29,8 @@
#include "butil/macros.h"
#include "butil/fd_utility.h"
#include "butil/debug/leak_annotations.h"
+#include "butil/memory/scope_guard.h"
+#include "butil/files/scoped_temp_dir.h"
#include <butil/fd_guard.h>
#include "bthread/countdown_event.h"
#include "bthread/unstable.h"
@@ -702,11 +704,11 @@ TEST_F(SocketTest, health_check) {
brpc::SocketId id = 8888;
butil::EndPoint point;
ASSERT_NO_FATAL_FAILURE(PickUnusedEndPoint(&point));
- const int kCheckInteval = 1;
+ const int kCheckInterval = 1;
brpc::SocketOptions options;
options.remote_side = point;
options.user = new CheckRecycle;
- options.health_check_interval_s = kCheckInteval/*s*/;
+ options.health_check_interval_s = kCheckInterval/*s*/;
ASSERT_EQ(0, brpc::Socket::Create(options, &id));
brpc::Socket* s = nullptr;
{
@@ -794,7 +796,7 @@ TEST_F(SocketTest, health_check) {
while (brpc::Socket::Status(id, &nref) != 0) {
bthread_usleep(1000);
ASSERT_LT(butil::cpuwide_time_us(),
- start_time + kCheckInteval * 1000000L + 100000L/*100ms*/);
+ start_time + kCheckInterval * 1000000L + 100000L/*100ms*/);
}
//ASSERT_EQ(2, nref);
ASSERT_TRUE(global_sock);
@@ -856,7 +858,7 @@ static void DoNothingOnEdgeTriggeredEvents(brpc::Socket*) {}
// and get dispatched within that window. `OnInputEvent` must drop such a
// stale event instead of crashing.
TEST_F(SocketTest, input_event_on_revived_socket) {
- const int kCheckInteval = 1;
+ const int kCheckInterval = 1;
butil::EndPoint point;
butil::fd_guard listening_fd;
ASSERT_NO_FATAL_FAILURE(ListenOnFreePort(&point, &listening_fd));
@@ -864,7 +866,7 @@ TEST_F(SocketTest, input_event_on_revived_socket) {
brpc::SocketId id = 8888;
brpc::SocketOptions options;
options.remote_side = point;
- options.health_check_interval_s = kCheckInteval;
+ options.health_check_interval_s = kCheckInterval;
// `OnInputEvent` returns early without an edge-triggered handler.
options.on_edge_triggered_events = DoNothingOnEdgeTriggeredEvents;
ASSERT_EQ(0, brpc::Socket::Create(options, &id));
@@ -897,7 +899,7 @@ TEST_F(SocketTest, input_event_on_revived_socket) {
while (brpc::Socket::Status(id) != 0) {
bthread_usleep(1000);
ASSERT_LT(butil::cpuwide_time_us(),
- start_time + kCheckInteval * 1000000L + 1000000L);
+ start_time + kCheckInterval * 1000000L + 1000000L);
}
ASSERT_EQ(-1, s->fd());
@@ -930,6 +932,189 @@ TEST_F(SocketTest, input_event_on_revived_socket) {
ASSERT_EQ(0, brpc::Socket::SetFailed(id));
}
+TEST_F(SocketTest, client_binding_endpoint_types) {
+ const char* inputs[] = {"127.0.0.1:12345", "[::1]:12345",
+ "unix:client-binding-test.sock"};
+ const char* normalized[] = {"127.0.0.1:0", "[::1]:0",
+ "unix:client-binding-test.sock"};
+ for (size_t i = 0; i < 3; ++i) {
+ SCOPED_TRACE(inputs[i]);
+ brpc::SocketOptions options;
+ // No connection is made: verify propagation without requiring a real
+ // network device or platform support for SO_BINDTODEVICE.
+ options.device_name = "test-device";
+ ASSERT_EQ(0, butil::str2endpoint(inputs[i], &options.local_side));
+ butil::EndPoint expected;
+ ASSERT_EQ(0, butil::str2endpoint(normalized[i], &expected));
+ brpc::SocketId id;
+ ASSERT_EQ(0, brpc::Socket::Create(options, &id));
+ BRPC_SCOPE_EXIT { brpc::Socket::SetFailed(id); };
+ brpc::SocketUniquePtr ptr;
+ ASSERT_EQ(0, brpc::Socket::Address(id, &ptr));
+ ASSERT_EQ(expected, ptr->_bind_local_side);
+ ASSERT_EQ(options.device_name, ptr->_device_name);
+ // Normalization must not mutate a shared extended endpoint.
+ ASSERT_STREQ(inputs[i],
butil::endpoint2str(options.local_side).c_str());
+
+ brpc::SocketUniquePtr short_socket;
+ ASSERT_EQ(0, ptr->GetShortSocket(&short_socket));
+ BRPC_SCOPE_EXIT { short_socket->SetFailed(); };
+ ASSERT_EQ(expected, short_socket->_bind_local_side);
+ ASSERT_EQ(options.device_name, short_socket->_device_name);
+ brpc::SocketUniquePtr pooled_socket;
+ ASSERT_EQ(0, ptr->GetPooledSocket(&pooled_socket));
+ BRPC_SCOPE_EXIT { pooled_socket->SetFailed(); };
+ ASSERT_EQ(expected, pooled_socket->_bind_local_side);
+ ASSERT_EQ(options.device_name, pooled_socket->_device_name);
+
+ ASSERT_EQ(0, ptr->SetFailed());
+ ptr->_is_hc_related_ref_held = true;
+ BRPC_SCOPE_EXIT { ptr->_is_hc_related_ref_held = false; };
+ ASSERT_EQ(0, ptr->WaitAndReset(1));
+ ASSERT_EQ(butil::EndPoint(), ptr->local_side());
+ ASSERT_EQ(expected, ptr->_bind_local_side);
+ ASSERT_EQ(options.device_name, ptr->_device_name);
+ }
+}
+
+TEST_F(SocketTest, client_binding_ipv6_connect) {
+ butil::EndPoint point;
+ ASSERT_EQ(0, butil::str2endpoint("[::1]:0", &point));
+ butil::fd_guard listening_fd(butil::tcp_listen(point));
+ ASSERT_GE(listening_fd, 0) << berror();
+ ASSERT_EQ(0, butil::get_local_side(listening_fd, &point));
+ brpc::SocketOptions options;
+ options.remote_side = point;
+ // This port is already occupied by the listener. Binding must use port 0.
+ options.local_side = point;
+ brpc::SocketId id;
+ ASSERT_EQ(0, brpc::Socket::Create(options, &id));
+ BRPC_SCOPE_EXIT { brpc::Socket::SetFailed(id); };
+ brpc::SocketUniquePtr ptr;
+ ASSERT_EQ(0, brpc::Socket::Address(id, &ptr));
+ for (int i = 0; i < 2; ++i) {
+ const timespec deadline = butil::milliseconds_from_now(1000);
+ butil::fd_guard fd(ptr->Connect(&deadline, nullptr, nullptr));
+ ASSERT_GE(fd, 0) << berror();
+ butil::EndPoint actual;
+ ASSERT_EQ(0, butil::get_local_side(fd, &actual));
+ sockaddr_storage addr{};
+ ASSERT_EQ(0, butil::endpoint2sockaddr(actual, &addr));
+ ASSERT_EQ(AF_INET6, addr.ss_family);
+ const sockaddr_in6* in6 = reinterpret_cast<const sockaddr_in6*>(&addr);
+ ASSERT_TRUE(IN6_IS_ADDR_LOOPBACK(&in6->sin6_addr));
+ ASSERT_NE(0, in6->sin6_port);
+ }
+}
+
+TEST_F(SocketTest, client_binding_uds_connect) {
+ butil::ScopedTempDir dir;
+ ASSERT_TRUE(dir.CreateUniqueTempDir());
+ // Deliberately use different path lengths for bind() and connect().
+ const std::string server = "unix:" +
dir.path().Append("server-long.sock").value();
+ const std::string client = "unix:" + dir.path().Append("c.sock").value();
+ brpc::SocketOptions options;
+ ASSERT_EQ(0, butil::str2endpoint(server.c_str(), &options.remote_side));
+ ASSERT_EQ(0, butil::str2endpoint(client.c_str(), &options.local_side));
+ butil::fd_guard listening_fd(butil::tcp_listen(options.remote_side));
+ ASSERT_GE(listening_fd, 0) << berror();
+ brpc::SocketId id;
+ ASSERT_EQ(0, brpc::Socket::Create(options, &id));
+ BRPC_SCOPE_EXIT { brpc::Socket::SetFailed(id); };
+ brpc::SocketUniquePtr ptr;
+ ASSERT_EQ(0, brpc::Socket::Address(id, &ptr));
+ const timespec deadline = butil::milliseconds_from_now(1000);
+ butil::fd_guard fd(ptr->Connect(&deadline, nullptr, nullptr));
+ ASSERT_GE(fd, 0) << berror();
+ butil::EndPoint actual;
+ ASSERT_EQ(0, butil::get_local_side(fd, &actual));
+ ASSERT_EQ(AF_UNIX, butil::get_endpoint_type(actual));
+ ASSERT_STREQ(client.c_str(), butil::endpoint2str(actual).c_str());
+}
+
+#if defined(OS_LINUX)
+TEST_F(SocketTest, keep_client_bind_after_revive) {
+ int kCheckInterval = 1;
+ butil::EndPoint point;
+ butil::fd_guard listening_fd;
+ ASSERT_NO_FATAL_FAILURE(ListenOnFreePort(&point, &listening_fd));
+
+ butil::EndPoint bind_point;
+ // A distinct loopback source address (127.0.0.2) is used so that
+ // getsockname() can tell whether the explicit bind() actually happened:
the
+ // kernel would pick 127.0.0.1 as the source for a 127.0.0.1 destination
when
+ // no bind() is performed. This relies on the whole 127/8 being loopback,
+ // which is Linux specific.
+ ASSERT_EQ(0, str2endpoint("127.0.0.2:0", &bind_point));
+
+ brpc::SocketId id = 8888;
+ brpc::SocketOptions options;
+ options.remote_side = point;
+ // The explicitly configured client source address
(ChannelOptions::client_host).
+ options.local_side = bind_point;
+ options.health_check_interval_s = kCheckInterval;
+ options.on_edge_triggered_events = DoNothingOnEdgeTriggeredEvents;
+ ASSERT_EQ(0, brpc::Socket::Create(options, &id));
+
+ brpc::Socket* s = nullptr;
+ {
+ // `WaitAndReset' inside the health check waits until nref drops back
+ // to 2, thus no SocketUniquePtr may be held across `SetFailed'.
+ brpc::SocketUniquePtr ptr;
+ ASSERT_EQ(0, brpc::Socket::Address(id, &ptr));
+ s = ptr.get();
+ }
+ ASSERT_EQ(-1, s->fd());
+
+ // First connection: must bind to the configured source IP.
+ butil::IOBuf src;
+ src.append("hello");
+ ASSERT_EQ(0, s->Write(&src));
+ int64_t start_time = butil::cpuwide_time_us();
+ while (s->fd() < 0) {
+ bthread_usleep(1000);
+ ASSERT_LT(butil::cpuwide_time_us(), start_time + 1000000L);
+ }
+ ASSERT_EQ(bind_point.ip, s->local_side().ip);
+
+ // Fail the Socket and wait for the health check to revive it. This runs
+ // `WaitAndReset` which clears the runtime endpoint, not the binding
config.
+ ASSERT_EQ(0, s->SetFailed());
+ start_time = butil::cpuwide_time_us();
+ while (brpc::Socket::Status(id) != 0) {
+ bthread_usleep(1000);
+ ASSERT_LT(butil::cpuwide_time_us(),
+ start_time + kCheckInterval * 1000000L + 1000000L);
+ }
+
+ ASSERT_EQ(butil::EndPoint(), s->local_side());
+ ASSERT_EQ(bind_point, s->_bind_local_side);
+
+ // Reconnect on demand and verify the explicit source IP is still bound.
+ butil::EndPoint local_after_revive;
+ {
+ brpc::SocketUniquePtr ptr;
+ ASSERT_EQ(0, brpc::Socket::Address(id, &ptr));
+ butil::IOBuf src2;
+ src2.append("world");
+ ASSERT_EQ(0, ptr->Write(&src2));
+ start_time = butil::cpuwide_time_us();
+ while (ptr->fd() < 0) {
+ bthread_usleep(1000);
+ ASSERT_LT(butil::cpuwide_time_us(), start_time + 1000000L);
+ }
+ local_after_revive = ptr->local_side();
+ }
+ ASSERT_EQ(bind_point.ip, local_after_revive.ip);
+
+ s->ReleaseHCRelatedReference();
+ // Must close the listening fd before SetFailed, otherwise the health
+ // check still has chance to get reconnected and revive the id.
+ listening_fd.reset(-1);
+ ASSERT_EQ(0, brpc::Socket::SetFailed(id));
+}
+#endif // OS_LINUX
+
void* Writer(void* void_arg) {
WriterArg* arg = static_cast<WriterArg*>(void_arg);
brpc::SocketUniquePtr sock;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
