From 62e5667270f3d404c92f8bc51087890ea0e11e15 Mon Sep 17 00:00:00 2001 From: chenBright Date: Mon, 12 Oct 2026 22:47:14 +0800 Subject: [PATCH] Fix flaky endpoint tests and misleading core backtraces --- test/endpoint_unittest.cpp | 247 ++++++++++++++++++++++++------------- test/run_tests.sh | 34 +++-- 2 files changed, 185 insertions(+), 96 deletions(-) diff --git a/test/endpoint_unittest.cpp b/test/endpoint_unittest.cpp index 3d293cd7f4..336faacc9c 100644 --- a/test/endpoint_unittest.cpp +++ b/test/endpoint_unittest.cpp @@ -16,19 +16,27 @@ // under the License. #include +#include +#include +#include #include #include #include "butil/errno.h" #include "butil/endpoint.h" #include "butil/logging.h" +#include "butil/memory/scope_guard.h" #include "butil/containers/flat_map.h" #include "butil/details/extended_endpoint.hpp" #include -#if defined(OS_MACOSX) +#if defined(OS_LINUX) +#include +#elif defined(OS_MACOSX) +#include #include #endif namespace butil { +int pthread_fd_wait(int fd, unsigned events, const timespec* abstime); int pthread_timed_connect(int sockfd, const struct sockaddr* serv_addr, socklen_t addrlen, const timespec* abstime); } @@ -500,119 +508,180 @@ 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. + // Drain each connection before starting the next to keep the queue empty. { - 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 accepted(accept(listener, nullptr, nullptr)); + ASSERT_LE(0, accepted) << "errno=" << errno; } { - butil::fd_guard sockfd(butil::tcp_connect(ep1, nullptr, 1000)); + butil::fd_guard sockfd(butil::tcp_connect(ep, nullptr, 1000)); 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)); + butil::fd_guard accepted(accept(listener, nullptr, nullptr)); + ASSERT_LE(0, accepted) << "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(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); - const int rc = butil::pthread_timed_connect( - sockfd, (struct sockaddr*) &serv_addr, - serv_addr_size, &abstime); - ASSERT_EQ(-1, rc); - ASSERT_EQ(ETIMEDOUT, errno); - 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(1); + ASSERT_EQ(0, butil::pthread_timed_connect( + sockfd, reinterpret_cast(&serv_addr), serv_addr_size, + timed ? &deadline : nullptr)) << "errno=" << errno; + ASSERT_EQ(blocking, butil::is_blocking(sockfd)); + ASSERT_EQ(0, butil::is_connected(sockfd)); + butil::fd_guard accepted(accept(listener, nullptr, nullptr)); + ASSERT_LE(0, accepted) << "errno=" << errno; + } } } -bool g_connect_startd = false; - -void TestConnectInterruptImpl(bool timed) { - butil::EndPoint ep; - ASSERT_EQ(0, butil::hostname2endpoint(g_hostname1, 80, &ep)); +#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 encoding used by pthread_timed_connect. + return butil::pthread_fd_wait( + fd, static_cast(EVFILT_READ), abstime); +#endif +} - 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)); +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]); + // An empty pipe tests the deadline without TCP queue or network assumptions. + timespec deadline = butil::milliseconds_from_now(300); + ASSERT_EQ(-1, WaitForPipe(reader, &deadline)); + ASSERT_EQ(ETIMEDOUT, errno); } -void* ConnectThread(void* arg) { - bool timed = *(bool*)arg; - TestConnectInterruptImpl(timed); - return nullptr; +#if defined(OS_MACOSX) +TEST(EndPointTest, fd_wait_kqueue_events) { + int fds[2]; + ASSERT_EQ(0, pipe(fds)); + butil::fd_guard reader(fds[0]); + butil::fd_guard writer(fds[1]); + timespec deadline = butil::seconds_from_now(5); + ASSERT_EQ(0, butil::pthread_fd_wait( + writer, static_cast(EVFILT_WRITE), &deadline)); + + const char byte = 'x'; + ASSERT_EQ(1, write(writer, &byte, sizeof(byte))); + ASSERT_EQ(0, butil::pthread_fd_wait( + reader, static_cast(EVFILT_READ), &deadline)); + + ASSERT_EQ(-1, butil::pthread_fd_wait(reader, 0u, &deadline)); + EXPECT_EQ(EINVAL, errno); } +#endif -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 FDWaitArgs { + int fd; + bool timed; + std::atomic started{false}; + std::atomic finished{false}; + int signal_mask_error = 0; + int result = -1; + int error = 0; +}; + +void* FDWaitThread(void* arg) { + FDWaitArgs* args = static_cast(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 TestFDWaitWithSignals(bool timed) { + int fds[2]; + ASSERT_EQ(0, pipe(fds)); + butil::fd_guard reader(fds[0]); + butil::fd_guard writer(fds[1]); + FDWaitArgs 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)); + // Exercise waiting with repeated signals before making the pipe readable. + // This checks signal delivery and the wait result, not an exact EINTR count. + for (int i = 0; i < 10 && + !args.finished.load(std::memory_order_acquire) && + butil::cpuwide_time_us() < deadline; ++i) { + const int rc = pthread_kill(tid, SIGURG); + EXPECT_EQ(0, rc); + if (rc != 0) { + break; + } + 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"; + // The SIGURG handler sets the flag; read it after joining the waiter. + EXPECT_EQ(1, g_sigurg_received); + EXPECT_EQ(0, args.result) << "errno=" << args.error; } -TEST(EndPointTest, interrupt) { - register_sigurg(); - TestConnectInterrupt(false); - TestConnectInterrupt(true); +TEST(EndPointTest, fd_wait_with_signals) { + 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 { + EXPECT_EQ(0, sigaction(SIGURG, &old_action, nullptr)); + }; + ASSERT_NO_FATAL_FAILURE(TestFDWaitWithSignals(false)); + ASSERT_NO_FATAL_FAILURE(TestFDWaitWithSignals(true)); } +#endif // defined(OS_LINUX) || defined(OS_MACOSX) } // end of namespace diff --git a/test/run_tests.sh b/test/run_tests.sh index dd80afd94f..daebf6f4a3 100755 --- a/test/run_tests.sh +++ b/test/run_tests.sh @@ -24,13 +24,17 @@ failed_test="" rc=0 test_bins="test_butil test_bvar bthread*unittest brpc*unittest" export ASAN_OPTIONS="detect_leaks=1:detect_stack_use_after_return=1" +# Keep the original stdin on fd 3 for asynchronous test processes. for test_bin in $test_bins; do test_num=$((test_num + 1)) >&2 echo "[runtest] $test_bin" # A passing death test may leave a core behind. Keep only cores from the # binary about to run so a later failure is symbolized against its core. rm -f core.* - ./$test_bin + # Preserve stdin while starting the test asynchronously to capture its PID. + ./"$test_bin" <&3 3<&- & + test_pid=$! + wait "$test_pid" # If ASan abort without detailed call stack of new/delete, # try to disable fast_unwind_on_malloc, which would be a performance killer. # ASAN_OPTIONS="fast_unwind_on_malloc=0:detect_leaks=0" ./$test_bin @@ -39,28 +43,44 @@ for test_bin in $test_bins; do failed_test="$test_bin" break; fi -done +done 3<&0 if [ $test_num -eq 0 ]; then >&2 echo "[runtest] Cannot find any tests" exit 1 fi print_bt () { - # find newest core file - COREFILE=$(find . -name "core*" -type f -printf "%T@ %p\n" | sort -k 1 -n | cut -d' ' -f 2- | tail -n 1) + case "${2-}" in + ''|*[!0-9]*) return ;; + esac + if [ "$2" -le 0 ]; then + return + fi + # CI uses core.%e.%p; also accept core.%p. Only symbolize a core from the + # failed test process; a passing death test has a different PID. + # Match the current-directory cleanup above; ignore nested stale cores. + COREFILE= + for candidate in ./core.*."$2" ./core."$2"; do + if [ ! -f "$candidate" ] || [ -L "$candidate" ]; then + continue + fi + if [ -z "$COREFILE" ] || [ "$candidate" -nt "$COREFILE" ]; then + COREFILE=$candidate + fi + done if [ ! -z "$COREFILE" ]; then >&2 echo "corefile=$COREFILE prog=$1" - gdb -c "$COREFILE" $1 -ex "bt" -ex "thread apply all bt" -ex "set pagination 0" -batch; + gdb -c "$COREFILE" "$1" -ex "bt" -ex "thread apply all bt" -ex "set pagination 0" -batch; fi } if [ -z "$failed_test" ]; then >&2 echo "[runtest] $test_num succeeded" elif [ $test_num -gt 1 ]; then - print_bt $failed_test + print_bt "$failed_test" "$test_pid" >&2 echo "[runtest] '$failed_test' failed, $((test_num-1)) succeeded" else - print_bt $failed_test + print_bt "$failed_test" "$test_pid" >&2 echo "[runtest] '$failed_test' failed" fi exit $rc