Fix flaky endpoint tests and misleading core backtraces

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

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