Copilot commented on code in PR #3544:
URL: https://github.com/apache/brpc/pull/3544#discussion_r4013586153
##########
src/brpc/socket.cpp:
##########
@@ -1293,19 +1312,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) {
+ if (::bind(sockfd, (struct sockaddr*)&cli_addr, cli_addr_size) != 0) {
PLOG(ERROR) << "Fail to bind client socket, errno=" <<
strerror(errno);
Review Comment:
`PLOG(...)` already appends the errno message; additionally streaming
`strerror(errno)` while labeling it as `errno=` is misleading (it prints the
string, not the numeric errno). Prefer either relying on `PLOG(ERROR) <<
\"...\";` alone, or logging `errno` numerically with the message string (e.g.,
`errno=<num> (<strerror>)`).
##########
test/brpc_socket_unittest.cpp:
##########
@@ -929,6 +931,182 @@ 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;
+ 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);
+ // 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);
+ 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(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);
+ }
+}
+
+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 kCheckInteval = 1;
Review Comment:
Typo in constant name: `kCheckInteval` should be `kCheckInterval` (and
ideally `const int` since it’s used as a constant). This improves readability
and reduces the chance of copy/paste mistakes.
##########
test/brpc_socket_unittest.cpp:
##########
@@ -929,6 +931,182 @@ 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;
+ 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);
+ // 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);
+ 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(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);
+ }
+}
+
+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 kCheckInteval = 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 = kCheckInteval;
Review Comment:
Typo in constant name: `kCheckInteval` should be `kCheckInterval` (and
ideally `const int` since it’s used as a constant). This improves readability
and reduces the chance of copy/paste mistakes.
##########
src/brpc/socket.cpp:
##########
@@ -2796,7 +2816,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;
Review Comment:
New behavior propagates device binding (and the preserved bind endpoint) to
pooled sub-sockets, but the added tests only assert `_bind_local_side`
propagation. Add/extend a unit test to validate `device_name` propagation as
well (e.g., create a socket with `options.device_name` set, call
`GetPooledSocket()`/`GetShortSocket()`, and assert the sub-socket uses the same
device binding policy).
##########
src/brpc/socket.cpp:
##########
@@ -796,6 +797,23 @@ int Socket::OnCreated(const SocketOptions& options) {
_tcp_user_timeout_ms = options.tcp_user_timeout_ms;
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");
+ 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");
Review Comment:
The failure reasons here are hard to debug and the chosen error codes are
questionable (`ENOMEM` for a conversion failure is likely inaccurate). Consider
including the original endpoint in the message (e.g.,
`endpoint2str(options.local_side)`), and use a more appropriate error code
(commonly `EINVAL`) or propagate a more specific error when available.
--
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]
