Copilot commented on code in PR #3599:
URL: https://github.com/apache/brpc/pull/3599#discussion_r4259751707
##########
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);
}
Review Comment:
These manual forward declarations can silently drift from the real
definitions (or conflict with a different signature on some platforms), which
would be hard to diagnose in tests. Prefer including the header that declares
these APIs (or adding a small internal test header) so the compiler enforces
signature consistency.
##########
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
+}
Review Comment:
On macOS, `EVFILT_READ`/`EVFILT_WRITE` are negative filter values. Casting
them to `unsigned` and relying on later conversion back to a signed type is
implementation-defined and can break depending on how `pthread_fd_wait`
interprets `events`. A safer approach is to keep the event type signed
end-to-end on the kqueue path (e.g., adjust `pthread_fd_wait` to accept an
`int`/`int16_t` filter on macOS, or introduce a macOS-only wrapper with a
signed parameter that `pthread_fd_wait` consumes without unsigned
round-tripping).
##########
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:
This comment is inaccurate: `g_sigurg_received` is written by the `SIGURG`
handler (signal context), not by the waiter thread’s normal execution. Please
update the comment to reflect that the handler sets the flag and it’s only read
after `pthread_join` to avoid races.
##########
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:
Restricting symbolization to cores named with the failing test process PID
avoids stale death-test cores, but it can also hide useful crash data when a
failing test produces a core for a *child* process (e.g., a death test that
actually crashes). Consider adding a fallback path when no
`core.*.$pid`/`core.$pid` exists: e.g., optionally symbolize the newest
`core.*` produced during this test invocation (bounded by a captured start
timestamp) and emit a clear message indicating it’s a fallback to avoid
confusion.
--
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]
