Shutdown fd

src/main.cpp

@@ -12,6 +12,7 @@
 #include <netdb.h>
 #include <netinet/tcp.h>
 #include <sys/epoll.h>
+#include <sys/eventfd.h>
 #include <sys/socket.h>
 #include <sys/types.h>
 #include <sys/un.h>
@@ -19,8 +20,8 @@
 #include <unistd.h>
 #include <vector>
 
-std::atomic<bool> shutdown_requested{false};
 std::atomic<int> activeConnections{0};
+int shutdown_eventfd = -1;
 
 #ifndef __has_feature
 #define __has_feature(x) 0
@@ -28,7 +29,12 @@ std::atomic<int> 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<std::thread> 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<int64_t> 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<struct epoll_event> 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<Connection> conn{
                   static_cast<Connection *>(events[i].data.ptr)};
@@ -355,18 +382,72 @@ 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)) {
+      // Create dedicated epoll instance for accept thread
-        struct sockaddr_storage addr;
+      int accept_epollfd = epoll_create1(EPOLL_CLOEXEC);
-        int fd = getAcceptFd(sockfd, &addr);
+      if (accept_epollfd == -1) {
-        if (fd == -1) {
+        perror("epoll_create1");
-          if (errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK) {
+        return;
-            // TODO revisit
-            std::this_thread::sleep_for(std::chrono::milliseconds(10));
-            continue;
       }
-          perror("accept4");
+
+      // 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("epoll_wait");
+          break;
+        }
+
+        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)
@@ -377,20 +458,28 @@ int main(int argc, char *argv[]) {
                 close(fd);
                 continue;
               }
+
               auto conn = std::make_unique<Connection>(
-            addr, fd, connectionId.fetch_add(1, std::memory_order_relaxed));
+                  addr, fd,
-        // Transfer new connection to epoll ownership
+                  connectionId.fetch_add(1, std::memory_order_relaxed));
+
+              // Transfer new connection to network thread epoll
               struct epoll_event event{};
-        event.events = EPOLLIN | EPOLLONESHOT |
+              event.events = EPOLLIN | EPOLLONESHOT | EPOLLRDHUP;
-                       EPOLLRDHUP; // Listen for reads and disconnects
               conn->tsan_release();
-        event.data.ptr = conn.release(); // epoll now owns the connection
+              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");
-          abort(); // Process termination - OS cleans up leaked connection
+                // 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;
 }