This is an automated email from the ASF dual-hosted git repository.
wwbmmm 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 5cef29d5 Fix flaky endpoint tests and misleading core backtraces
(#3599)
5cef29d5 is described below
commit 5cef29d5dd55c0db93cec8ab8a62c4c3b524dce3
Author: Bright Chen <[email protected]>
AuthorDate: Sun Oct 18 10:32:01 2026 +0800
Fix flaky endpoint tests and misleading core backtraces (#3599)
---
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 3d293cd7..336faacc 100644
--- a/test/endpoint_unittest.cpp
+++ b/test/endpoint_unittest.cpp
@@ -16,19 +16,27 @@
// under the License.
#include <gtest/gtest.h>
+#include <atomic>
+#include <pthread.h>
+#include <signal.h>
#include <butil/fd_guard.h>
#include <butil/fd_utility.h>
#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 <netinet/tcp.h>
-#if defined(OS_MACOSX)
+#if defined(OS_LINUX)
+#include <sys/epoll.h>
+#elif defined(OS_MACOSX)
+#include <sys/event.h>
#include <netinet/tcp_fsm.h>
#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<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));
+ 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<unsigned>(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<unsigned>(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<unsigned>(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<bool> started{false};
+ std::atomic<bool> finished{false};
+ int signal_mask_error = 0;
+ int result = -1;
+ int error = 0;
+};
+
+void* FDWaitThread(void* arg) {
+ FDWaitArgs* args = static_cast<FDWaitArgs*>(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 dd80afd9..daebf6f4 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
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
