chenBright commented on code in PR #3599:
URL: https://github.com/apache/brpc/pull/3599#discussion_r4258353071
##########
test/endpoint_unittest.cpp:
##########
@@ -500,119 +508,213 @@ 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));
+ 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));
+ }
}
+}
+#if defined(OS_LINUX)
+TEST(EndPointTest, tcp_connect_timeout_with_full_accept_queue) {
+ 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;
+ // Linux allows one queued connection with backlog=0. Keep it unaccepted
+ // so subsequent handshakes must wait instead of relying on Internet
latency.
+ ASSERT_EQ(0, listen(listener, 0));
+ ASSERT_EQ(0, butil::get_local_side(listener, &ep));
+ butil::fd_guard queued(butil::tcp_connect(ep, nullptr, 1000));
+ ASSERT_LE(0, queued) << "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(butil::tcp_connect(ep, nullptr, 300));
+ const int error = errno;
+ ASSERT_EQ(-1, sockfd);
+ // tcp_abort_on_overflow may reject the connection rather than time
out.
+ ASSERT_TRUE(error == ETIMEDOUT || error == ECONNREFUSED)
+ << "errno=" << error;
+ }
+ sockaddr_storage serv_addr{};
+ socklen_t serv_addr_size = 0;
+ ASSERT_EQ(0, endpoint2sockaddr(ep, &serv_addr, &serv_addr_size));
+ for (bool blocking : {false, true}) {
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);
+ if (!blocking) {
+ ASSERT_EQ(0, butil::make_non_blocking(sockfd));
+ }
+ timespec abstime = butil::milliseconds_from_now(300);
const int rc = butil::pthread_timed_connect(
- sockfd, (struct sockaddr*) &serv_addr,
+ sockfd, reinterpret_cast<sockaddr*>(&serv_addr),
serv_addr_size, &abstime);
+ const int error = errno;
ASSERT_EQ(-1, rc);
- ASSERT_EQ(ETIMEDOUT, errno);
- ASSERT_EQ(is_blocking, butil::is_blocking(sockfd));
+ ASSERT_TRUE(error == ETIMEDOUT || error == ECONNREFUSED)
+ << "errno=" << error;
+ ASSERT_EQ(blocking, butil::is_blocking(sockfd));
}
}
+#endif
-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]);
+ 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
+
+volatile sig_atomic_t g_sigurg_received = 0;
-void do_nothing_handler(int) {}
+void RecordSigurg(int) {
+ g_sigurg_received = 1;
+}
-void register_sigurg() {
- signal(SIGURG, do_nothing_handler);
+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 TestConnectInterrupt(bool timed) {
- g_connect_startd = false;
+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.
+ 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 {
Review Comment:
The explicitly included butil/memory/scope_guard.h directly defines
BRPC_SCOPE_EXIT at line 103. The macro uses butil’s scope-guard implementation
and introduces no additional brpc dependency. The test compiles successfully.
--
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]
