From 8ef118a17d8f9e21f9a2c834b4e9469886093a9c Mon Sep 17 00:00:00 2001 From: Andrew Noyes Date: Mon, 18 Aug 2025 15:21:45 -0400 Subject: [PATCH] Shutdown fd --- src/main.cpp | 166 ++++++++++++++++++++++++++++++++++++++++----------- 1 file changed, 130 insertions(+), 36 deletions(-) diff --git a/src/main.cpp b/src/main.cpp index f1e413e..06fba1c 100644 --- a/src/main.cpp +++ b/src/main.cpp @@ -12,6 +12,7 @@ #include #include #include +#include #include #include #include @@ -19,8 +20,8 @@ #include #include -std::atomic shutdown_requested{false}; std::atomic activeConnections{0}; +int shutdown_eventfd = -1; #ifndef __has_feature #define __has_feature(x) 0 @@ -28,7 +29,12 @@ std::atomic activeConnections{0}; void signal_handler(int sig) { if (sig == SIGTERM || sig == SIGINT) { - shutdown_requested.store(true, std::memory_order_relaxed); + if (shutdown_eventfd != -1) { + uint64_t val = 1; + if (write(shutdown_eventfd, &val, sizeof(val)) == -1) { + abort(); // Critical failure - can't signal shutdown + } + } } } @@ -205,6 +211,9 @@ struct Connection { return false; } + // This is necessary because tsan doesn't (yet?) understand that there's a + // happens-before relationship for epoll_ctl(..., EPOLL_CTL_MOD, ...) and + // epoll_wait #if __has_feature(thread_sanitizer) void tsan_acquire() { tsan_sync.load(std::memory_order_acquire); } void tsan_release() { tsan_sync.store(0, std::memory_order_release); } @@ -255,6 +264,13 @@ int main(int argc, char *argv[]) { << config->subscription.keepalive_interval.count() << " seconds" << std::endl; + // Create shutdown eventfd for graceful shutdown + shutdown_eventfd = eventfd(0, EFD_CLOEXEC); + if (shutdown_eventfd == -1) { + perror("eventfd"); + return 1; + } + signal(SIGPIPE, SIG_IGN); signal(SIGTERM, signal_handler); signal(SIGINT, signal_handler); @@ -262,11 +278,16 @@ int main(int argc, char *argv[]) { int sockfd = getListenFd(config->server.bind_address.c_str(), std::to_string(config->server.port).c_str()); std::vector threads; - int epollfd = epoll_create(/*ignored*/ 1); + int epollfd = epoll_create1(EPOLL_CLOEXEC); if (epollfd == -1) { perror("epoll_create"); abort(); } + // Add shutdown eventfd to network thread epoll + struct epoll_event shutdown_event; + shutdown_event.events = EPOLLIN; + shutdown_event.data.fd = shutdown_eventfd; + epoll_ctl(epollfd, EPOLL_CTL_ADD, shutdown_eventfd, &shutdown_event); std::atomic connectionId{0}; @@ -280,12 +301,12 @@ int main(int argc, char *argv[]) { event_batch_size = config->server.event_batch_size]() { pthread_setname_np(pthread_self(), ("network-" + std::to_string(i)).c_str()); - while (!shutdown_requested.load(std::memory_order_relaxed)) { + while (true) { std::vector events(event_batch_size); int eventCount; for (;;) { eventCount = epoll_wait(epollfd, events.data(), event_batch_size, - 1000 /* 1 second timeout */); + -1 /* no timeout */); if (eventCount == -1) { if (errno == EINTR) { continue; @@ -302,6 +323,12 @@ int main(int argc, char *argv[]) { } for (int i = 0; i < eventCount; ++i) { + // Check for shutdown event + if (events[i].data.fd == shutdown_eventfd) { + // Don't read - let other threads see it too + return; + } + // Take ownership from epoll: raw pointer -> unique_ptr std::unique_ptr conn{ static_cast(events[i].data.ptr)}; @@ -355,42 +382,104 @@ int main(int argc, char *argv[]) { max_connections = config->server.max_connections]() { pthread_setname_np(pthread_self(), ("accept-" + std::to_string(i)).c_str()); - // Call accept in a loop - while (!shutdown_requested.load(std::memory_order_relaxed)) { - struct sockaddr_storage addr; - int fd = getAcceptFd(sockfd, &addr); - if (fd == -1) { - if (errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK) { - // TODO revisit - std::this_thread::sleep_for(std::chrono::milliseconds(10)); + + // Create dedicated epoll instance for accept thread + int accept_epollfd = epoll_create1(EPOLL_CLOEXEC); + if (accept_epollfd == -1) { + perror("epoll_create1"); + return; + } + + // Add listen socket to accept epoll + struct epoll_event listen_event; + listen_event.events = EPOLLIN; + listen_event.data.fd = sockfd; + if (epoll_ctl(accept_epollfd, EPOLL_CTL_ADD, sockfd, &listen_event) == + -1) { + perror("epoll_ctl listen socket"); + close(accept_epollfd); + return; + } + + // Add shutdown eventfd to accept epoll + struct epoll_event shutdown_event; + shutdown_event.events = EPOLLIN; + shutdown_event.data.fd = shutdown_eventfd; + if (epoll_ctl(accept_epollfd, EPOLL_CTL_ADD, shutdown_eventfd, + &shutdown_event) == -1) { + perror("epoll_ctl shutdown eventfd"); + close(accept_epollfd); + return; + } + + while (true) { + struct epoll_event events[2]; // listen socket + shutdown eventfd + int ready = epoll_wait(accept_epollfd, events, 2, -1 /* no timeout */); + + if (ready == -1) { + if (errno == EINTR) continue; - } - perror("accept4"); - continue; + perror("epoll_wait"); + break; } - // Check connection limit (0 means unlimited) - if (max_connections > 0 && - activeConnections.load(std::memory_order_relaxed) >= - max_connections) { - // Reject connection by immediately closing it - close(fd); - continue; - } - auto conn = std::make_unique( - addr, fd, connectionId.fetch_add(1, std::memory_order_relaxed)); - // Transfer new connection to epoll ownership - struct epoll_event event{}; - event.events = EPOLLIN | EPOLLONESHOT | - EPOLLRDHUP; // Listen for reads and disconnects - conn->tsan_release(); - event.data.ptr = conn.release(); // epoll now owns the connection - int e = epoll_ctl(epollfd, EPOLL_CTL_ADD, fd, &event); - if (e == -1) { - perror("epoll_ctl"); - abort(); // Process termination - OS cleans up leaked connection + if (ready == 0) + continue; // Timeout - check shutdown flag + + for (int j = 0; j < ready; ++j) { + if (events[j].data.fd == shutdown_eventfd) { + // Don't read - let other threads see it too + close(accept_epollfd); + return; + } + + if (events[j].data.fd == sockfd) { + // Listen socket ready - accept connections + while (true) { + struct sockaddr_storage addr; + socklen_t addrlen = sizeof(addr); + int fd = accept4(sockfd, (struct sockaddr *)&addr, &addrlen, + SOCK_NONBLOCK); + + if (fd == -1) { + if (errno == EAGAIN || errno == EWOULDBLOCK) + break; // No more connections + if (errno == EINTR) + continue; + perror("accept4"); + break; + } + + // Check connection limit (0 means unlimited) + if (max_connections > 0 && + activeConnections.load(std::memory_order_relaxed) >= + max_connections) { + // Reject connection by immediately closing it + close(fd); + continue; + } + + auto conn = std::make_unique( + addr, fd, + connectionId.fetch_add(1, std::memory_order_relaxed)); + + // Transfer new connection to network thread epoll + struct epoll_event event{}; + event.events = EPOLLIN | EPOLLONESHOT | EPOLLRDHUP; + conn->tsan_release(); + event.data.ptr = + conn.release(); // network epoll now owns the connection + int e = epoll_ctl(epollfd, EPOLL_CTL_ADD, fd, &event); + if (e == -1) { + perror("epoll_ctl"); + // TODO: Better error handling - connection will be leaked + } + } + } } } + + close(accept_epollfd); }); } @@ -398,5 +487,10 @@ int main(int argc, char *argv[]) { t.join(); } + // Cleanup + close(shutdown_eventfd); + close(epollfd); + close(sockfd); + return 0; }