weaselab/weaseldb
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

More error handling

Browse Source 
Handle epoll_ctl errors by closing the connection
This commit is contained in:
Andrew Noyes
2025-08-18 15:59:35 -04:00
parent 141ae823ab
commit 8b18c5fde9
1 changed files with 67 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files

129
src/main.cpp
View File

@@ -301,72 +301,75 @@ int main(int argc, char *argv[]) {
int networkThreads = config->server.network_threads; int networkThreads = config->server.network_threads;
for (int i = 0; i < networkThreads; ++i) { for (int i = 0; i < networkThreads; ++i) {
threads.emplace_back([network_epollfd, i, threads.emplace_back(
max_request_size = [network_epollfd, i,
config->server.max_request_size_bytes, max_request_size = config->server.max_request_size_bytes,
event_batch_size = event_batch_size = config->server.event_batch_size]() {
config->server.event_batch_size]() { pthread_setname_np(pthread_self(),
pthread_setname_np(pthread_self(), ("network-" + std::to_string(i)).c_str());
("network-" + std::to_string(i)).c_str()); std::vector<struct epoll_event> events(event_batch_size);
std::vector<struct epoll_event> events(event_batch_size); for (;;) {
for (;;) { int eventCount = epoll_wait(network_epollfd, events.data(),
int eventCount = epoll_wait(network_epollfd, events.data(), event_batch_size, -1 /* no timeout */);
event_batch_size, -1 /* no timeout */); if (eventCount == -1) {
if (eventCount == -1) { if (errno == EINTR) {
if (errno == EINTR) { continue;
continue; }
} perror("epoll_wait");
perror("epoll_wait"); abort();
abort(); }
}
for (int i = 0; i < eventCount; ++i) { for (int i = 0; i < eventCount; ++i) {
// Check for shutdown event // Check for shutdown event
if (events[i].data.fd == shutdown_eventfd) { if (events[i].data.fd == shutdown_eventfd) {
// Don't read eventfd - all threads need to see shutdown signal // Don't read eventfd - all threads need to see shutdown signal
return; return;
} }
// Take ownership from epoll: raw pointer -> unique_ptr // Take ownership from epoll: raw pointer -> unique_ptr
std::unique_ptr<Connection> conn{ std::unique_ptr<Connection> conn{
static_cast<Connection *>(events[i].data.ptr)}; static_cast<Connection *>(events[i].data.ptr)};
conn->tsan_acquire(); conn->tsan_acquire();
events[i].data.ptr = nullptr; // Clear epoll pointer (we own it now) events[i].data.ptr =
const int fd = conn->fd; nullptr; // Clear epoll pointer (we own it now)
const int fd = conn->fd;
if (events[i].events & (EPOLLERR | EPOLLHUP | EPOLLRDHUP)) { if (events[i].events & (EPOLLERR | EPOLLHUP | EPOLLRDHUP)) {
// Connection closed or error occurred - unique_ptr destructor // Connection closed or error occurred - unique_ptr destructor
// cleans up // cleans up
continue; continue;
} }
if (events[i].events & EPOLLIN) { if (events[i].events & EPOLLIN) {
conn->readBytes(max_request_size); conn->readBytes(max_request_size);
} }
if (events[i].events & EPOLLOUT) { if (events[i].events & EPOLLOUT) {
bool done = conn->writeBytes(); bool done = conn->writeBytes();
if (done) { if (done) {
continue; continue;
}
}
if (conn->tasks.empty()) {
events[i].events = EPOLLIN | EPOLLONESHOT | EPOLLRDHUP;
} else {
events[i].events = EPOLLOUT | EPOLLONESHOT | EPOLLRDHUP;
}
// Transfer ownership back to epoll: unique_ptr -> raw pointer
conn->tsan_release();
Connection *raw_conn =
conn.release(); // Get raw pointer before epoll_ctl
events[i].data.ptr = raw_conn; // epoll now owns the connection
int e = epoll_ctl(network_epollfd, EPOLL_CTL_MOD, fd, &events[i]);
if (e == -1) {
perror("epoll_ctl");
delete raw_conn; // Clean up connection on epoll failure
continue;
}
} }
} }
});
if (conn->tasks.empty()) {
events[i].events = EPOLLIN | EPOLLONESHOT | EPOLLRDHUP;
} else {
events[i].events = EPOLLOUT | EPOLLONESHOT | EPOLLRDHUP;
}
// Transfer ownership back to epoll: unique_ptr -> raw pointer
conn->tsan_release();
events[i].data.ptr = conn.release(); // epoll now owns the connection
int e = epoll_ctl(network_epollfd, EPOLL_CTL_MOD, fd, &events[i]);
if (e == -1) {
perror("epoll_ctl");
abort(); // Connection leaked on abort() is acceptable - OS cleanup
}
}
}
});
} }
// Accept threads from configuration // Accept threads from configuration
@@ -454,13 +457,15 @@ int main(int argc, char *argv[]) {
struct epoll_event event{}; struct epoll_event event{};
event.events = EPOLLIN | EPOLLONESHOT | EPOLLRDHUP; event.events = EPOLLIN | EPOLLONESHOT | EPOLLRDHUP;
conn->tsan_release(); conn->tsan_release();
Connection *raw_conn =
conn.release(); // Get raw pointer before epoll_ctl
event.data.ptr = event.data.ptr =
conn.release(); // network epoll now owns the connection raw_conn; // network epoll now owns the connection
int e = epoll_ctl(network_epollfd, EPOLL_CTL_ADD, fd, &event); int e = epoll_ctl(network_epollfd, EPOLL_CTL_ADD, fd, &event);
if (e == -1) { if (e == -1) {
perror("epoll_ctl"); perror("epoll_ctl");
abort(); // Connection leaked on abort() is acceptable - OS delete raw_conn; // Clean up connection on epoll failure
// cleanup continue;
} }
} }
} }