chenBright commented on code in PR #3599:
URL: https://github.com/apache/brpc/pull/3599#discussion_r4260133883
##########
test/run_tests.sh:
##########
@@ -39,28 +43,35 @@ 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.
+ COREFILE=$(find . \( -name "core.*.$2" -o -name "core.$2" \) -type f
-printf "%T@ %p\n" | sort -k 1 -n | cut -d' ' -f 2- | tail -n 1)
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
}
Review Comment:
Declined. A passing death test can generate a child core during the same
invocation as an unrelated assertion failure. Filtering by start time would
still select that unrelated core and reintroduce the original misleading crash
report. Keep the parent-PID filter.
##########
test/endpoint_unittest.cpp:
##########
@@ -500,119 +508,172 @@ 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, 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));
- }
- {
- 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));
+ }
}
}
-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
+ 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";
+ // Only the waiter writes this flag; read it after joining the thread.
Review Comment:
Fixed. The comment now explicitly identifies the SIGURG handler as the
writer. The flag is read after joining the waiter.
--
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]
