Prepare for more per io thread/epoll instance state

src/connection.cpp

@@ -1,7 +1,6 @@
 #include "connection.hpp"
 
 #include <cerrno>
-#include <climits>
 #include <cstdio>
 #include <cstdlib>
 #include <sys/epoll.h>
@@ -119,7 +118,8 @@ void Connection::append_bytes(std::span<std::string_view> data_parts,
       // I think we have to call epoll_ctl while holding mutex_. Otherwise a
       // call that clears the write interest could get reordered with one that
       // sets it and we would hang.
-      epoll_ctl(server->epoll_fds_[epoll_index_], EPOLL_CTL_MOD, fd_, &event);
+      epoll_ctl(server->event_loops_[epoll_index_].epoll_fd_, EPOLL_CTL_MOD,
+                fd_, &event);
     }
   }
 }
@@ -147,7 +147,8 @@ void Connection::send_response(void *protocol_context,
       event.data.fd = fd_;
       event.events = EPOLLIN | EPOLLOUT;
       tsan_release();
-      epoll_ctl(server->epoll_fds_[epoll_index_], EPOLL_CTL_MOD, fd_, &event);
+      epoll_ctl(server->event_loops_[epoll_index_].epoll_fd_, EPOLL_CTL_MOD,
+                fd_, &event);
     }
   }
 }
@@ -296,7 +297,8 @@ uint32_t Connection::write_bytes() {
         // I think we have to call epoll_ctl while holding mutex_. Otherwise a
         // call that clears the write interest could get reordered with one that
         // sets it and we would hang.
-        epoll_ctl(server->epoll_fds_[epoll_index_], EPOLL_CTL_MOD, fd_, &event);
+        epoll_ctl(server->event_loops_[epoll_index_].epoll_fd_, EPOLL_CTL_MOD,
+                  fd_, &event);
       }
       // Handle shutdown modes after all messages are sent
       if (shutdown_requested_ == ConnectionShutdown::WriteOnly) {

src/server.cpp

@@ -1,6 +1,5 @@
 #include "server.hpp"
 
-#include <csignal>
 #include <cstdio>
 #include <cstdlib>
 #include <cstring>
@@ -75,7 +74,7 @@ Server::~Server() {
   }
 
   // Close all epoll instances
-  for (int epollfd : epoll_fds_) {
+  for (auto [epollfd] : event_loops_) {
     if (epollfd != -1) {
       int e = close(epollfd);
       if (e == -1 && errno != EINTR) {
@@ -84,7 +83,7 @@ Server::~Server() {
       }
     }
   }
-  epoll_fds_.clear();
+  event_loops_.clear();
 
   // Close all listen sockets (Server always owns them)
   for (int fd : listen_fds_) {
@@ -167,7 +166,7 @@ int Server::create_local_connection() {
   // Use round-robin distribution for local connections across epoll instances
   size_t epoll_index =
       connection_distribution_counter_.fetch_add(1, std::memory_order_relaxed) %
-      epoll_fds_.size();
+      event_loops_.size();
 
   // Create Connection object
   auto connection = make_ref<Connection>(
@@ -184,7 +183,7 @@ int Server::create_local_connection() {
   event.events = EPOLLIN;
   event.data.fd = server_fd;
 
-  int epollfd = epoll_fds_[epoll_index];
+  int epollfd = event_loops_[epoll_index].epoll_fd_;
   if (epoll_ctl(epollfd, EPOLL_CTL_ADD, server_fd, &event) == -1) {
     perror("epoll_ctl ADD local connection");
     connection_registry_.remove(server_fd);
@@ -221,11 +220,11 @@ void Server::setup_shutdown_pipe() {
 void Server::create_epoll_instances() {
   // Create one epoll instance per I/O thread (1:1 mapping) to eliminate
   // contention
-  epoll_fds_.resize(config_.server.io_threads);
+  event_loops_.resize(config_.server.io_threads);
 
   for (int i = 0; i < config_.server.io_threads; ++i) {
-    epoll_fds_[i] = epoll_create1(EPOLL_CLOEXEC);
-    if (epoll_fds_[i] == -1) {
+    event_loops_[i].epoll_fd_ = epoll_create1(EPOLL_CLOEXEC);
+    if (event_loops_[i].epoll_fd_ == -1) {
       perror("epoll_create1");
       std::abort();
     }
@@ -235,7 +234,7 @@ void Server::create_epoll_instances() {
     shutdown_event.events = EPOLLIN;
     shutdown_event.data.fd = shutdown_pipe_[0];
 
-    if (epoll_ctl(epoll_fds_[i], EPOLL_CTL_ADD, shutdown_pipe_[0],
+    if (epoll_ctl(event_loops_[i].epoll_fd_, EPOLL_CTL_ADD, shutdown_pipe_[0],
                   &shutdown_event) == -1) {
       perror("epoll_ctl shutdown pipe");
       std::abort();
@@ -247,8 +246,8 @@ void Server::create_epoll_instances() {
       struct epoll_event listen_event;
       listen_event.events = EPOLLIN | EPOLLEXCLUSIVE;
       listen_event.data.fd = listen_fd;
-      if (epoll_ctl(epoll_fds_[i], EPOLL_CTL_ADD, listen_fd, &listen_event) ==
-          -1) {
+      if (epoll_ctl(event_loops_[i].epoll_fd_, EPOLL_CTL_ADD, listen_fd,
+                    &listen_event) == -1) {
         perror("epoll_ctl listen socket");
         std::abort();
       }
@@ -265,7 +264,7 @@ void Server::start_io_threads(std::vector<std::thread> &threads) {
                          ("io-" + std::to_string(thread_id)).c_str());
 
       // Each thread uses its assigned epoll instance (1:1 mapping)
-      int epollfd = epoll_fds_[thread_id];
+      int epollfd = event_loops_[thread_id].epoll_fd_;
 
       std::vector<epoll_event> events(config_.server.event_batch_size);
       std::vector<Ref<Connection>> batch(config_.server.event_batch_size);

src/server.hpp

@@ -131,8 +131,12 @@ private:
   // Shutdown coordination
   int shutdown_pipe_[2] = {-1, -1};
 
+  struct EventLoopState {
+    int epoll_fd_;
+  };
+
   // Multiple epoll file descriptors (1:1 with I/O threads) to reduce contention
-  std::vector<int> epoll_fds_;
+  std::vector<EventLoopState> event_loops_;
   std::vector<int>
       listen_fds_; // FDs to accept connections on (Server owns these)