Copilot commented on code in PR #3599:
URL: https://github.com/apache/brpc/pull/3599#discussion_r4247576989
##########
test/endpoint_unittest.cpp:
##########
@@ -500,119 +508,188 @@ TEST(EndPointTest, endpoint_concurrency) {
}
}
-const char* g_hostname1 = "github.com";
-const char* g_hostname2 = "baidu.com";
TEST(EndPointTest, tcp_connect) {
- butil::EndPoint ep1;
- butil::EndPoint ep2;
- ASSERT_EQ(0, butil::hostname2endpoint(g_hostname1, 80, &ep1));
- ASSERT_EQ(0, butil::hostname2endpoint(g_hostname2, 80, &ep2));
+ butil::EndPoint ep;
+ ASSERT_EQ(0, butil::str2endpoint("127.0.0.1:0", &ep));
+ butil::fd_guard listener(butil::tcp_listen(ep));
+ ASSERT_LE(0, listener) << "errno=" << errno;
+ ASSERT_EQ(0, butil::get_local_side(listener, &ep));
+ // The local TCP handshake completes without a concurrent accept thread.
{
- butil::fd_guard sockfd(butil::tcp_connect(ep1, nullptr));
+ butil::fd_guard sockfd(butil::tcp_connect(ep, nullptr));
ASSERT_LE(0, sockfd) << "errno=" << errno;
}
{
- butil::fd_guard sockfd(butil::tcp_connect(ep1, nullptr, 1000));
+ butil::fd_guard sockfd(butil::tcp_connect(ep, nullptr, 5000));
ASSERT_LE(0, sockfd) << "errno=" << errno;
}
- {
- butil::fd_guard sockfd(butil::tcp_connect(ep2, nullptr, 1));
- ASSERT_EQ(-1, sockfd) << "errno=" << errno;
- ASSERT_EQ(ETIMEDOUT, errno);
- }
- {
- struct sockaddr_storage serv_addr{};
- socklen_t serv_addr_size = 0;
- ASSERT_EQ(0, endpoint2sockaddr(ep1, &serv_addr, &serv_addr_size));
- butil::fd_guard sockfd(socket(serv_addr.ss_family, SOCK_STREAM, 0));
- ASSERT_LE(0, sockfd);
- bool is_blocking = butil::is_blocking(sockfd);
- ASSERT_EQ(0, butil::pthread_timed_connect(
- sockfd, (struct sockaddr*) &serv_addr, serv_addr_size, nullptr));
- ASSERT_EQ(is_blocking, butil::is_blocking(sockfd));
+ sockaddr_storage serv_addr{};
+ socklen_t serv_addr_size = 0;
+ ASSERT_EQ(0, endpoint2sockaddr(ep, &serv_addr, &serv_addr_size));
+ for (bool timed : {false, true}) {
+ for (bool blocking : {false, true}) {
+ butil::fd_guard sockfd(socket(serv_addr.ss_family, SOCK_STREAM,
0));
+ ASSERT_LE(0, sockfd);
+ if (!blocking) {
+ ASSERT_EQ(0, butil::make_non_blocking(sockfd));
+ }
+ timespec deadline = butil::seconds_from_now(5);
+ ASSERT_EQ(0, butil::pthread_timed_connect(
+ sockfd, reinterpret_cast<sockaddr*>(&serv_addr),
serv_addr_size,
+ timed ? &deadline : nullptr)) << "errno=" << errno;
+ ASSERT_EQ(blocking, butil::is_blocking(sockfd));
+ ASSERT_EQ(0, butil::is_connected(sockfd));
+ }
}
+}
+#if defined(OS_LINUX)
+TEST(EndPointTest, tcp_connect_timeout_with_full_accept_queue) {
+ butil::EndPoint ep;
+ ASSERT_EQ(0, butil::str2endpoint("127.0.0.1:0", &ep));
+ butil::fd_guard listener(butil::tcp_listen(ep));
+ ASSERT_LE(0, listener);
+ // Linux allows one queued connection with backlog=0. Keep it unaccepted
+ // so subsequent handshakes must wait instead of relying on Internet
latency.
+ ASSERT_EQ(0, listen(listener, 0));
+ ASSERT_EQ(0, butil::get_local_side(listener, &ep));
+ butil::fd_guard queued(butil::tcp_connect(ep, nullptr, 5000));
+ ASSERT_LE(0, queued) << "errno=" << errno;
{
- struct sockaddr_storage serv_addr{};
- socklen_t serv_addr_size = 0;
- ASSERT_EQ(0, endpoint2sockaddr(ep2, &serv_addr, &serv_addr_size));
+ butil::fd_guard sockfd(butil::tcp_connect(ep, nullptr, 50));
Review Comment:
Several new/updated checks rely on ~50ms timeouts. On heavily contended CI
machines, 50ms can be below typical scheduling jitter and cause intermittent
failures even for correct behavior. Increasing these to a slightly larger value
(e.g., a few hundred ms) would generally preserve test intent while improving
reliability.
##########
test/endpoint_unittest.cpp:
##########
@@ -500,119 +508,188 @@ TEST(EndPointTest, endpoint_concurrency) {
}
}
-const char* g_hostname1 = "github.com";
-const char* g_hostname2 = "baidu.com";
TEST(EndPointTest, tcp_connect) {
- butil::EndPoint ep1;
- butil::EndPoint ep2;
- ASSERT_EQ(0, butil::hostname2endpoint(g_hostname1, 80, &ep1));
- ASSERT_EQ(0, butil::hostname2endpoint(g_hostname2, 80, &ep2));
+ butil::EndPoint ep;
+ ASSERT_EQ(0, butil::str2endpoint("127.0.0.1:0", &ep));
+ butil::fd_guard listener(butil::tcp_listen(ep));
+ ASSERT_LE(0, listener) << "errno=" << errno;
+ ASSERT_EQ(0, butil::get_local_side(listener, &ep));
+ // The local TCP handshake completes without a concurrent accept thread.
{
- butil::fd_guard sockfd(butil::tcp_connect(ep1, nullptr));
+ butil::fd_guard sockfd(butil::tcp_connect(ep, nullptr));
ASSERT_LE(0, sockfd) << "errno=" << errno;
}
{
- butil::fd_guard sockfd(butil::tcp_connect(ep1, nullptr, 1000));
+ butil::fd_guard sockfd(butil::tcp_connect(ep, nullptr, 5000));
ASSERT_LE(0, sockfd) << "errno=" << errno;
}
- {
- butil::fd_guard sockfd(butil::tcp_connect(ep2, nullptr, 1));
- ASSERT_EQ(-1, sockfd) << "errno=" << errno;
- ASSERT_EQ(ETIMEDOUT, errno);
- }
- {
- struct sockaddr_storage serv_addr{};
- socklen_t serv_addr_size = 0;
- ASSERT_EQ(0, endpoint2sockaddr(ep1, &serv_addr, &serv_addr_size));
- butil::fd_guard sockfd(socket(serv_addr.ss_family, SOCK_STREAM, 0));
- ASSERT_LE(0, sockfd);
- bool is_blocking = butil::is_blocking(sockfd);
- ASSERT_EQ(0, butil::pthread_timed_connect(
- sockfd, (struct sockaddr*) &serv_addr, serv_addr_size, nullptr));
- ASSERT_EQ(is_blocking, butil::is_blocking(sockfd));
+ sockaddr_storage serv_addr{};
+ socklen_t serv_addr_size = 0;
+ ASSERT_EQ(0, endpoint2sockaddr(ep, &serv_addr, &serv_addr_size));
+ for (bool timed : {false, true}) {
+ for (bool blocking : {false, true}) {
+ butil::fd_guard sockfd(socket(serv_addr.ss_family, SOCK_STREAM,
0));
+ ASSERT_LE(0, sockfd);
+ if (!blocking) {
+ ASSERT_EQ(0, butil::make_non_blocking(sockfd));
+ }
+ timespec deadline = butil::seconds_from_now(5);
+ ASSERT_EQ(0, butil::pthread_timed_connect(
+ sockfd, reinterpret_cast<sockaddr*>(&serv_addr),
serv_addr_size,
+ timed ? &deadline : nullptr)) << "errno=" << errno;
+ ASSERT_EQ(blocking, butil::is_blocking(sockfd));
+ ASSERT_EQ(0, butil::is_connected(sockfd));
+ }
}
+}
+#if defined(OS_LINUX)
+TEST(EndPointTest, tcp_connect_timeout_with_full_accept_queue) {
+ butil::EndPoint ep;
+ ASSERT_EQ(0, butil::str2endpoint("127.0.0.1:0", &ep));
+ butil::fd_guard listener(butil::tcp_listen(ep));
+ ASSERT_LE(0, listener);
+ // Linux allows one queued connection with backlog=0. Keep it unaccepted
+ // so subsequent handshakes must wait instead of relying on Internet
latency.
+ ASSERT_EQ(0, listen(listener, 0));
+ ASSERT_EQ(0, butil::get_local_side(listener, &ep));
+ butil::fd_guard queued(butil::tcp_connect(ep, nullptr, 5000));
+ ASSERT_LE(0, queued) << "errno=" << errno;
{
- struct sockaddr_storage serv_addr{};
- socklen_t serv_addr_size = 0;
- ASSERT_EQ(0, endpoint2sockaddr(ep2, &serv_addr, &serv_addr_size));
+ butil::fd_guard sockfd(butil::tcp_connect(ep, nullptr, 50));
+ const int error = errno;
+ ASSERT_EQ(-1, sockfd);
+ // tcp_abort_on_overflow may reject the connection rather than time
out.
+ ASSERT_TRUE(error == ETIMEDOUT || error == ECONNREFUSED)
+ << "errno=" << error;
+ }
+ sockaddr_storage serv_addr{};
+ socklen_t serv_addr_size = 0;
+ ASSERT_EQ(0, endpoint2sockaddr(ep, &serv_addr, &serv_addr_size));
+ for (bool blocking : {false, true}) {
butil::fd_guard sockfd(socket(serv_addr.ss_family, SOCK_STREAM, 0));
ASSERT_LE(0, sockfd);
- bool is_blocking = butil::is_blocking(sockfd);
- // In most cases, 1 millisecond will result in a connection timeout.
- timespec abstime = butil::milliseconds_from_now(1);
+ if (!blocking) {
+ ASSERT_EQ(0, butil::make_non_blocking(sockfd));
+ }
+ timespec abstime = butil::milliseconds_from_now(50);
Review Comment:
Several new/updated checks rely on ~50ms timeouts. On heavily contended CI
machines, 50ms can be below typical scheduling jitter and cause intermittent
failures even for correct behavior. Increasing these to a slightly larger value
(e.g., a few hundred ms) would generally preserve test intent while improving
reliability.
##########
test/endpoint_unittest.cpp:
##########
@@ -500,119 +508,188 @@ TEST(EndPointTest, endpoint_concurrency) {
}
}
-const char* g_hostname1 = "github.com";
-const char* g_hostname2 = "baidu.com";
TEST(EndPointTest, tcp_connect) {
- butil::EndPoint ep1;
- butil::EndPoint ep2;
- ASSERT_EQ(0, butil::hostname2endpoint(g_hostname1, 80, &ep1));
- ASSERT_EQ(0, butil::hostname2endpoint(g_hostname2, 80, &ep2));
+ butil::EndPoint ep;
+ ASSERT_EQ(0, butil::str2endpoint("127.0.0.1:0", &ep));
+ butil::fd_guard listener(butil::tcp_listen(ep));
+ ASSERT_LE(0, listener) << "errno=" << errno;
+ ASSERT_EQ(0, butil::get_local_side(listener, &ep));
+ // The local TCP handshake completes without a concurrent accept thread.
{
- butil::fd_guard sockfd(butil::tcp_connect(ep1, nullptr));
+ butil::fd_guard sockfd(butil::tcp_connect(ep, nullptr));
ASSERT_LE(0, sockfd) << "errno=" << errno;
}
{
- butil::fd_guard sockfd(butil::tcp_connect(ep1, nullptr, 1000));
+ butil::fd_guard sockfd(butil::tcp_connect(ep, nullptr, 5000));
ASSERT_LE(0, sockfd) << "errno=" << errno;
}
- {
- butil::fd_guard sockfd(butil::tcp_connect(ep2, nullptr, 1));
- ASSERT_EQ(-1, sockfd) << "errno=" << errno;
- ASSERT_EQ(ETIMEDOUT, errno);
- }
- {
- struct sockaddr_storage serv_addr{};
- socklen_t serv_addr_size = 0;
- ASSERT_EQ(0, endpoint2sockaddr(ep1, &serv_addr, &serv_addr_size));
- butil::fd_guard sockfd(socket(serv_addr.ss_family, SOCK_STREAM, 0));
- ASSERT_LE(0, sockfd);
- bool is_blocking = butil::is_blocking(sockfd);
- ASSERT_EQ(0, butil::pthread_timed_connect(
- sockfd, (struct sockaddr*) &serv_addr, serv_addr_size, nullptr));
- ASSERT_EQ(is_blocking, butil::is_blocking(sockfd));
+ sockaddr_storage serv_addr{};
+ socklen_t serv_addr_size = 0;
+ ASSERT_EQ(0, endpoint2sockaddr(ep, &serv_addr, &serv_addr_size));
+ for (bool timed : {false, true}) {
+ for (bool blocking : {false, true}) {
+ butil::fd_guard sockfd(socket(serv_addr.ss_family, SOCK_STREAM,
0));
+ ASSERT_LE(0, sockfd);
+ if (!blocking) {
+ ASSERT_EQ(0, butil::make_non_blocking(sockfd));
+ }
+ timespec deadline = butil::seconds_from_now(5);
+ ASSERT_EQ(0, butil::pthread_timed_connect(
+ sockfd, reinterpret_cast<sockaddr*>(&serv_addr),
serv_addr_size,
+ timed ? &deadline : nullptr)) << "errno=" << errno;
+ ASSERT_EQ(blocking, butil::is_blocking(sockfd));
+ ASSERT_EQ(0, butil::is_connected(sockfd));
+ }
}
+}
+#if defined(OS_LINUX)
+TEST(EndPointTest, tcp_connect_timeout_with_full_accept_queue) {
+ butil::EndPoint ep;
+ ASSERT_EQ(0, butil::str2endpoint("127.0.0.1:0", &ep));
+ butil::fd_guard listener(butil::tcp_listen(ep));
+ ASSERT_LE(0, listener);
+ // Linux allows one queued connection with backlog=0. Keep it unaccepted
+ // so subsequent handshakes must wait instead of relying on Internet
latency.
+ ASSERT_EQ(0, listen(listener, 0));
+ ASSERT_EQ(0, butil::get_local_side(listener, &ep));
+ butil::fd_guard queued(butil::tcp_connect(ep, nullptr, 5000));
+ ASSERT_LE(0, queued) << "errno=" << errno;
{
- struct sockaddr_storage serv_addr{};
- socklen_t serv_addr_size = 0;
- ASSERT_EQ(0, endpoint2sockaddr(ep2, &serv_addr, &serv_addr_size));
+ butil::fd_guard sockfd(butil::tcp_connect(ep, nullptr, 50));
+ const int error = errno;
+ ASSERT_EQ(-1, sockfd);
+ // tcp_abort_on_overflow may reject the connection rather than time
out.
+ ASSERT_TRUE(error == ETIMEDOUT || error == ECONNREFUSED)
+ << "errno=" << error;
+ }
+ sockaddr_storage serv_addr{};
+ socklen_t serv_addr_size = 0;
+ ASSERT_EQ(0, endpoint2sockaddr(ep, &serv_addr, &serv_addr_size));
+ for (bool blocking : {false, true}) {
butil::fd_guard sockfd(socket(serv_addr.ss_family, SOCK_STREAM, 0));
ASSERT_LE(0, sockfd);
- bool is_blocking = butil::is_blocking(sockfd);
- // In most cases, 1 millisecond will result in a connection timeout.
- timespec abstime = butil::milliseconds_from_now(1);
+ if (!blocking) {
+ ASSERT_EQ(0, butil::make_non_blocking(sockfd));
+ }
+ timespec abstime = butil::milliseconds_from_now(50);
const int rc = butil::pthread_timed_connect(
- sockfd, (struct sockaddr*) &serv_addr,
+ sockfd, reinterpret_cast<sockaddr*>(&serv_addr),
serv_addr_size, &abstime);
+ const int error = errno;
ASSERT_EQ(-1, rc);
- ASSERT_EQ(ETIMEDOUT, errno);
- ASSERT_EQ(is_blocking, butil::is_blocking(sockfd));
+ ASSERT_TRUE(error == ETIMEDOUT || error == ECONNREFUSED)
+ << "errno=" << error;
+ ASSERT_EQ(blocking, butil::is_blocking(sockfd));
}
}
+#endif
-bool g_connect_startd = false;
-
-void TestConnectInterruptImpl(bool timed) {
- butil::EndPoint ep;
- ASSERT_EQ(0, butil::hostname2endpoint(g_hostname1, 80, &ep));
-
- struct sockaddr_storage serv_addr{};
- socklen_t serv_addr_size = 0;
- ASSERT_EQ(0, endpoint2sockaddr(ep, &serv_addr, &serv_addr_size));
- butil::fd_guard sockfd(socket(serv_addr.ss_family, SOCK_STREAM, 0));
- ASSERT_LE(0, sockfd);
-
- int rc;
- if (timed) {
- int64_t start_ms = butil::cpuwide_time_ms();
- butil::tcp_connect(ep, nullptr);
- int64_t connect_ms = butil::cpuwide_time_ms() - start_ms;
- LOG(INFO) << "Connect to " << ep << ", cost " << connect_ms << "ms";
-
- timespec abstime = butil::milliseconds_from_now(connect_ms * 10);
- rc = butil::pthread_timed_connect(
- sockfd, (struct sockaddr*) &serv_addr,
- serv_addr_size, &abstime);
- } else {
- rc = butil::pthread_timed_connect(
- sockfd, (struct sockaddr*) &serv_addr,
- serv_addr_size, nullptr);
- }
- ASSERT_EQ(0, rc) << "errno=" << errno;
- ASSERT_EQ(0, butil::is_connected(sockfd));
+#if defined(OS_LINUX) || defined(OS_MACOSX)
+int WaitForPipe(int fd, const timespec* abstime) {
+#if defined(OS_LINUX)
+ return butil::pthread_fd_wait(fd, EPOLLIN, abstime);
+#else
+ // Match the unsigned event interface used by pthread_fd_wait and its
+ // existing kqueue callers, including pthread_timed_connect.
+ return butil::pthread_fd_wait(fd, static_cast<unsigned>(EVFILT_READ),
abstime);
+#endif
}
-void* ConnectThread(void* arg) {
- bool timed = *(bool*)arg;
- TestConnectInterruptImpl(timed);
- return nullptr;
+TEST(EndPointTest, fd_wait_timeout) {
+ int fds[2];
+ ASSERT_EQ(0, pipe(fds));
+ butil::fd_guard reader(fds[0]);
+ butil::fd_guard writer(fds[1]);
+ timespec deadline = butil::milliseconds_from_now(50);
Review Comment:
Several new/updated checks rely on ~50ms timeouts. On heavily contended CI
machines, 50ms can be below typical scheduling jitter and cause intermittent
failures even for correct behavior. Increasing these to a slightly larger value
(e.g., a few hundred ms) would generally preserve test intent while improving
reliability.
##########
test/run_tests.sh:
##########
@@ -46,6 +46,14 @@ if [ $test_num -eq 0 ]; then
fi
print_bt () {
+ case "${2-}" in
+ ''|*[!0-9]*) return ;;
+ esac
+ # Assertion failures exit normally. A passing death test may have left a
+ # core in this binary, so only print a backtrace when it exited on a
signal.
+ if [ "$2" -le 128 ]; then
+ return
+ fi
Review Comment:
The current heuristic treats any numeric exit status > 128 as a signal exit,
but processes can also `exit(n)` with `n > 128`. That can reintroduce
misleading core backtraces for non-signal failures. Consider validating that
`($2 - 128)` maps to a real signal (e.g., via `kill -l`) before printing the
backtrace; otherwise return early.
##########
test/endpoint_unittest.cpp:
##########
@@ -500,119 +508,188 @@ TEST(EndPointTest, endpoint_concurrency) {
}
}
-const char* g_hostname1 = "github.com";
-const char* g_hostname2 = "baidu.com";
TEST(EndPointTest, tcp_connect) {
- butil::EndPoint ep1;
- butil::EndPoint ep2;
- ASSERT_EQ(0, butil::hostname2endpoint(g_hostname1, 80, &ep1));
- ASSERT_EQ(0, butil::hostname2endpoint(g_hostname2, 80, &ep2));
+ butil::EndPoint ep;
+ ASSERT_EQ(0, butil::str2endpoint("127.0.0.1:0", &ep));
+ butil::fd_guard listener(butil::tcp_listen(ep));
+ ASSERT_LE(0, listener) << "errno=" << errno;
+ ASSERT_EQ(0, butil::get_local_side(listener, &ep));
+ // The local TCP handshake completes without a concurrent accept thread.
{
- butil::fd_guard sockfd(butil::tcp_connect(ep1, nullptr));
+ butil::fd_guard sockfd(butil::tcp_connect(ep, nullptr));
ASSERT_LE(0, sockfd) << "errno=" << errno;
}
{
- butil::fd_guard sockfd(butil::tcp_connect(ep1, nullptr, 1000));
+ butil::fd_guard sockfd(butil::tcp_connect(ep, nullptr, 5000));
ASSERT_LE(0, sockfd) << "errno=" << errno;
}
- {
- butil::fd_guard sockfd(butil::tcp_connect(ep2, nullptr, 1));
- ASSERT_EQ(-1, sockfd) << "errno=" << errno;
- ASSERT_EQ(ETIMEDOUT, errno);
- }
- {
- struct sockaddr_storage serv_addr{};
- socklen_t serv_addr_size = 0;
- ASSERT_EQ(0, endpoint2sockaddr(ep1, &serv_addr, &serv_addr_size));
- butil::fd_guard sockfd(socket(serv_addr.ss_family, SOCK_STREAM, 0));
- ASSERT_LE(0, sockfd);
- bool is_blocking = butil::is_blocking(sockfd);
- ASSERT_EQ(0, butil::pthread_timed_connect(
- sockfd, (struct sockaddr*) &serv_addr, serv_addr_size, nullptr));
- ASSERT_EQ(is_blocking, butil::is_blocking(sockfd));
+ sockaddr_storage serv_addr{};
+ socklen_t serv_addr_size = 0;
+ ASSERT_EQ(0, endpoint2sockaddr(ep, &serv_addr, &serv_addr_size));
+ for (bool timed : {false, true}) {
+ for (bool blocking : {false, true}) {
+ butil::fd_guard sockfd(socket(serv_addr.ss_family, SOCK_STREAM,
0));
+ ASSERT_LE(0, sockfd);
+ if (!blocking) {
+ ASSERT_EQ(0, butil::make_non_blocking(sockfd));
+ }
+ timespec deadline = butil::seconds_from_now(5);
+ ASSERT_EQ(0, butil::pthread_timed_connect(
+ sockfd, reinterpret_cast<sockaddr*>(&serv_addr),
serv_addr_size,
+ timed ? &deadline : nullptr)) << "errno=" << errno;
+ ASSERT_EQ(blocking, butil::is_blocking(sockfd));
+ ASSERT_EQ(0, butil::is_connected(sockfd));
+ }
}
+}
+#if defined(OS_LINUX)
+TEST(EndPointTest, tcp_connect_timeout_with_full_accept_queue) {
+ butil::EndPoint ep;
+ ASSERT_EQ(0, butil::str2endpoint("127.0.0.1:0", &ep));
+ butil::fd_guard listener(butil::tcp_listen(ep));
+ ASSERT_LE(0, listener);
+ // Linux allows one queued connection with backlog=0. Keep it unaccepted
+ // so subsequent handshakes must wait instead of relying on Internet
latency.
+ ASSERT_EQ(0, listen(listener, 0));
+ ASSERT_EQ(0, butil::get_local_side(listener, &ep));
+ butil::fd_guard queued(butil::tcp_connect(ep, nullptr, 5000));
+ ASSERT_LE(0, queued) << "errno=" << errno;
{
- struct sockaddr_storage serv_addr{};
- socklen_t serv_addr_size = 0;
- ASSERT_EQ(0, endpoint2sockaddr(ep2, &serv_addr, &serv_addr_size));
+ butil::fd_guard sockfd(butil::tcp_connect(ep, nullptr, 50));
+ const int error = errno;
+ ASSERT_EQ(-1, sockfd);
+ // tcp_abort_on_overflow may reject the connection rather than time
out.
+ ASSERT_TRUE(error == ETIMEDOUT || error == ECONNREFUSED)
+ << "errno=" << error;
+ }
+ sockaddr_storage serv_addr{};
+ socklen_t serv_addr_size = 0;
+ ASSERT_EQ(0, endpoint2sockaddr(ep, &serv_addr, &serv_addr_size));
+ for (bool blocking : {false, true}) {
butil::fd_guard sockfd(socket(serv_addr.ss_family, SOCK_STREAM, 0));
ASSERT_LE(0, sockfd);
- bool is_blocking = butil::is_blocking(sockfd);
- // In most cases, 1 millisecond will result in a connection timeout.
- timespec abstime = butil::milliseconds_from_now(1);
+ if (!blocking) {
+ ASSERT_EQ(0, butil::make_non_blocking(sockfd));
+ }
+ timespec abstime = butil::milliseconds_from_now(50);
const int rc = butil::pthread_timed_connect(
- sockfd, (struct sockaddr*) &serv_addr,
+ sockfd, reinterpret_cast<sockaddr*>(&serv_addr),
serv_addr_size, &abstime);
+ const int error = errno;
ASSERT_EQ(-1, rc);
- ASSERT_EQ(ETIMEDOUT, errno);
- ASSERT_EQ(is_blocking, butil::is_blocking(sockfd));
+ ASSERT_TRUE(error == ETIMEDOUT || error == ECONNREFUSED)
+ << "errno=" << error;
+ ASSERT_EQ(blocking, butil::is_blocking(sockfd));
}
}
+#endif
-bool g_connect_startd = false;
-
-void TestConnectInterruptImpl(bool timed) {
- butil::EndPoint ep;
- ASSERT_EQ(0, butil::hostname2endpoint(g_hostname1, 80, &ep));
-
- struct sockaddr_storage serv_addr{};
- socklen_t serv_addr_size = 0;
- ASSERT_EQ(0, endpoint2sockaddr(ep, &serv_addr, &serv_addr_size));
- butil::fd_guard sockfd(socket(serv_addr.ss_family, SOCK_STREAM, 0));
- ASSERT_LE(0, sockfd);
-
- int rc;
- if (timed) {
- int64_t start_ms = butil::cpuwide_time_ms();
- butil::tcp_connect(ep, nullptr);
- int64_t connect_ms = butil::cpuwide_time_ms() - start_ms;
- LOG(INFO) << "Connect to " << ep << ", cost " << connect_ms << "ms";
-
- timespec abstime = butil::milliseconds_from_now(connect_ms * 10);
- rc = butil::pthread_timed_connect(
- sockfd, (struct sockaddr*) &serv_addr,
- serv_addr_size, &abstime);
- } else {
- rc = butil::pthread_timed_connect(
- sockfd, (struct sockaddr*) &serv_addr,
- serv_addr_size, nullptr);
- }
- ASSERT_EQ(0, rc) << "errno=" << errno;
- ASSERT_EQ(0, butil::is_connected(sockfd));
+#if defined(OS_LINUX) || defined(OS_MACOSX)
+int WaitForPipe(int fd, const timespec* abstime) {
+#if defined(OS_LINUX)
+ return butil::pthread_fd_wait(fd, EPOLLIN, abstime);
+#else
+ // Match the unsigned event interface used by pthread_fd_wait and its
+ // existing kqueue callers, including pthread_timed_connect.
+ return butil::pthread_fd_wait(fd, static_cast<unsigned>(EVFILT_READ),
abstime);
+#endif
}
-void* ConnectThread(void* arg) {
- bool timed = *(bool*)arg;
- TestConnectInterruptImpl(timed);
- return nullptr;
+TEST(EndPointTest, fd_wait_timeout) {
+ int fds[2];
+ ASSERT_EQ(0, pipe(fds));
+ butil::fd_guard reader(fds[0]);
+ butil::fd_guard writer(fds[1]);
+ timespec deadline = butil::milliseconds_from_now(50);
+ ASSERT_EQ(-1, WaitForPipe(reader, &deadline));
+ ASSERT_EQ(ETIMEDOUT, errno);
}
-void do_nothing_handler(int) {}
+volatile sig_atomic_t g_sigurg_received = 0;
-void register_sigurg() {
- signal(SIGURG, do_nothing_handler);
+void RecordSigurg(int) {
+ g_sigurg_received = 1;
}
-void TestConnectInterrupt(bool timed) {
- g_connect_startd = false;
+struct FDWaitInterruptArgs {
+ int fd;
+ bool timed;
+ std::atomic<bool> started{false};
+ std::atomic<bool> finished{false};
+ int signal_mask_error = 0;
+ int result = -1;
+ int error = 0;
+};
+
+void* FDWaitThread(void* arg) {
+ FDWaitInterruptArgs* args = static_cast<FDWaitInterruptArgs*>(arg);
+ sigset_t signals;
+ sigemptyset(&signals);
+ sigaddset(&signals, SIGURG);
+ args->signal_mask_error = pthread_sigmask(SIG_UNBLOCK, &signals, nullptr);
+ timespec deadline = butil::seconds_from_now(5);
+ args->started.store(true, std::memory_order_release);
+ if (args->signal_mask_error == 0) {
+ args->result = WaitForPipe(args->fd, args->timed ? &deadline :
nullptr);
+ if (args->result != 0) {
+ args->error = errno;
+ }
+ }
+ args->finished.store(true, std::memory_order_release);
+ return nullptr;
+}
+
+void TestFDWaitInterrupt(bool timed) {
+ int fds[2];
+ ASSERT_EQ(0, pipe(fds));
+ butil::fd_guard reader(fds[0]);
+ butil::fd_guard writer(fds[1]);
+ FDWaitInterruptArgs args;
+ args.fd = reader;
+ args.timed = timed;
+ g_sigurg_received = 0;
pthread_t tid;
- ASSERT_EQ(0, pthread_create(&tid, nullptr, ConnectThread, &timed));
+ ASSERT_EQ(0, pthread_create(&tid, nullptr, FDWaitThread, &args));
- while (g_connect_startd) {
+ const int64_t deadline = butil::cpuwide_time_us() + 5000000L;
+ while (!args.started.load(std::memory_order_acquire) &&
+ butil::cpuwide_time_us() < deadline) {
usleep(1000);
}
-
- ASSERT_EQ(0, pthread_kill(tid, SIGURG));
-
- pthread_join(tid, nullptr);
+ EXPECT_TRUE(args.started.load(std::memory_order_acquire));
+ // Repeated signals reach the waiter while poll is blocked on the empty
+ // pipe. EINTR must not end either the timed or the untimed wait.
+ for (int i = 0; i < 10 && !args.finished.load(std::memory_order_acquire);
++i) {
+ EXPECT_EQ(0, pthread_kill(tid, SIGURG));
+ usleep(1000);
+ }
+ EXPECT_FALSE(args.finished.load(std::memory_order_acquire));
+
+ const char byte = 'x';
+ EXPECT_EQ(1, write(writer, &byte, sizeof(byte)));
+ // Closing the writer also wakes poll if an earlier expectation failed.
+ writer.reset(-1);
+ ASSERT_EQ(0, pthread_join(tid, nullptr));
+ ASSERT_EQ(0, args.signal_mask_error) << "pthread_sigmask failed";
+ // Only the waiter writes this flag; read it after joining the thread.
+ EXPECT_EQ(1, g_sigurg_received);
+ EXPECT_EQ(0, args.result) << "errno=" << args.error;
}
TEST(EndPointTest, interrupt) {
- register_sigurg();
- TestConnectInterrupt(false);
- TestConnectInterrupt(true);
+ struct sigaction action{};
+ struct sigaction old_action;
+ action.sa_handler = RecordSigurg;
+ sigemptyset(&action.sa_mask);
+ ASSERT_EQ(0, sigaction(SIGURG, &action, &old_action));
+ BRPC_SCOPE_EXIT {
+ CHECK_EQ(0, sigaction(SIGURG, &old_action, nullptr));
Review Comment:
Using `CHECK_EQ` in a unit test cleanup path can abort the entire test
process (and mask the original failure), which is especially undesirable in a
test intended to reduce misleading failures. Prefer reporting a gtest failure
instead (e.g., `ADD_FAILURE()` / `EXPECT_EQ`) while still attempting to restore
the handler.
--
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]
