diff --git a/src/connection.cpp b/src/connection.cpp index 1d452a0..0652965 100644 --- a/src/connection.cpp +++ b/src/connection.cpp @@ -1,7 +1,6 @@ #include "connection.hpp" #include -#include #include #include #include @@ -119,7 +118,8 @@ void Connection::append_bytes(std::span 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) { diff --git a/src/server.cpp b/src/server.cpp index ae68e5c..ab0036e 100644 --- a/src/server.cpp +++ b/src/server.cpp @@ -1,6 +1,5 @@ #include "server.hpp" -#include #include #include #include @@ -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( @@ -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 &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 events(config_.server.event_batch_size); std::vector> batch(config_.server.event_batch_size); diff --git a/src/server.hpp b/src/server.hpp index 618f2b6..ab4fc8a 100644 --- a/src/server.hpp +++ b/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 epoll_fds_; + std::vector event_loops_; std::vector listen_fds_; // FDs to accept connections on (Server owns these)