net: add Linux epoll fdwatch backend

2026-04-13 20:32:06 +02:00
parent bf72bd2669
commit 54a4134bf7
3 changed files with 359 additions and 55 deletions
--- a/src/libthecore/fdwatch.cpp
+++ b/src/libthecore/fdwatch.cpp
@@ -1,6 +1,6 @@
 #include "stdafx.h"

-#ifndef __USE_SELECT__
+#if defined(OS_FREEBSD)

 LPFDWATCH fdwatch_new(int nfiles)
 {
@@ -230,7 +230,307 @@ void * fdwatch_get_client_data(LPFDWATCH fdw, unsigned int event_idx)

    return (fdw->fd_data[fd]);
 }
-#else	// ifndef __USE_SELECT__
+
+#elif defined(__linux__)
+
+#ifndef EPOLLRDHUP
+#define EPOLLRDHUP 0
+#endif
+
+static int fdwatch_interest_mask(int rw)
+{
+	return rw & (FDW_READ | FDW_WRITE);
+}
+
+static uint32_t fdwatch_build_epoll_events(int rw)
+{
+	uint32_t events = 0;
+
+	if (rw & FDW_READ)
+		events |= EPOLLIN | EPOLLRDHUP;
+
+	if (rw & FDW_WRITE)
+		events |= EPOLLOUT;
+
+	return events;
+}
+
+static int fdwatch_epoll_ctl(LPFDWATCH fdw, int op, socket_t fd)
+{
+	struct epoll_event event;
+
+	memset(&event, 0, sizeof(event));
+	event.events = fdwatch_build_epoll_events(fdw->fd_rw[fd]);
+	event.data.fd = fd;
+
+	if (op == EPOLL_CTL_DEL)
+		return epoll_ctl(fdw->epfd, op, fd, NULL);
+
+	return epoll_ctl(fdw->epfd, op, fd, &event);
+}
+
+static bool fdwatch_update_registration(LPFDWATCH fdw, socket_t fd, int old_rw)
+{
+	const int old_interest = fdwatch_interest_mask(old_rw);
+	const int new_interest = fdwatch_interest_mask(fdw->fd_rw[fd]);
+
+	if (!old_interest && !new_interest)
+		return true;
+
+	if (!new_interest)
+	{
+		if (fdwatch_epoll_ctl(fdw, EPOLL_CTL_DEL, fd) == 0 || errno == ENOENT)
+			return true;
+
+		return false;
+	}
+
+	if (!old_interest)
+	{
+		if (fdwatch_epoll_ctl(fdw, EPOLL_CTL_ADD, fd) == 0)
+			return true;
+
+		if (errno == EEXIST && fdwatch_epoll_ctl(fdw, EPOLL_CTL_MOD, fd) == 0)
+			return true;
+
+		return false;
+	}
+
+	if (fdwatch_epoll_ctl(fdw, EPOLL_CTL_MOD, fd) == 0)
+		return true;
+
+	if (errno == ENOENT && fdwatch_epoll_ctl(fdw, EPOLL_CTL_ADD, fd) == 0)
+		return true;
+
+	return false;
+}
+
+LPFDWATCH fdwatch_new(int nfiles)
+{
+	LPFDWATCH fdw;
+	int epfd;
+
+	epfd = epoll_create1(EPOLL_CLOEXEC);
+
+	if (epfd == -1)
+	{
+		sys_err("%s", strerror(errno));
+		return NULL;
+	}
+
+	CREATE(fdw, FDWATCH, 1);
+
+	fdw->epfd = epfd;
+	fdw->nfiles = nfiles;
+
+	CREATE(fdw->ep_events, EPOLL_EVENT, nfiles);
+	CREATE(fdw->ready_events, EPOLL_EVENT, nfiles * 2);
+	CREATE(fdw->fd_rw, int, nfiles);
+	CREATE(fdw->fd_data, void*, nfiles);
+
+	return (fdw);
+}
+
+void fdwatch_clear_fd(LPFDWATCH fdw, socket_t fd)
+{
+	if (fd < 0 || fd >= fdw->nfiles)
+		return;
+
+	fdw->fd_data[fd] = NULL;
+	fdw->fd_rw[fd] = 0;
+}
+
+void fdwatch_delete(LPFDWATCH fdw)
+{
+	close(fdw->epfd);
+	free(fdw->fd_data);
+	free(fdw->fd_rw);
+	free(fdw->ep_events);
+	free(fdw->ready_events);
+	free(fdw);
+}
+
+int fdwatch_check_fd(LPFDWATCH fdw, socket_t fd)
+{
+	return 0;
+}
+
+void fdwatch_add_fd(LPFDWATCH fdw, socket_t fd, void* client_data, int rw, int oneshot)
+{
+	int old_rw;
+	void* old_data;
+
+	if (fd < 0 || fd >= fdw->nfiles)
+	{
+		sys_err("fd overflow %d", fd);
+		return;
+	}
+
+	old_rw = fdw->fd_rw[fd];
+
+	if (old_rw & rw)
+		return;
+
+	old_data = fdw->fd_data[fd];
+
+	fdw->fd_rw[fd] |= rw;
+
+	if (rw & FDW_WRITE)
+	{
+		if (oneshot)
+			fdw->fd_rw[fd] |= FDW_WRITE_ONESHOT;
+		else
+			fdw->fd_rw[fd] &= ~FDW_WRITE_ONESHOT;
+	}
+
+	fdw->fd_data[fd] = client_data;
+
+	if (!fdwatch_update_registration(fdw, fd, old_rw))
+	{
+		sys_err("epoll_ctl failed for fd %d [%d] %s", fd, errno, strerror(errno));
+		fdw->fd_rw[fd] = old_rw;
+		fdw->fd_data[fd] = old_data;
+	}
+}
+
+void fdwatch_del_fd(LPFDWATCH fdw, socket_t fd)
+{
+	int old_rw;
+
+	if (fd < 0 || fd >= fdw->nfiles)
+		return;
+
+	old_rw = fdw->fd_rw[fd];
+
+	if (!fdwatch_interest_mask(old_rw) && !fdw->fd_data[fd])
+		return;
+
+	fdwatch_clear_fd(fdw, fd);
+
+	if (!fdwatch_update_registration(fdw, fd, old_rw))
+		sys_err("epoll_ctl delete failed for fd %d [%d] %s", fd, errno, strerror(errno));
+}
+
+int fdwatch(LPFDWATCH fdw, struct timeval *timeout)
+{
+	int ready;
+	int timeout_ms;
+	int event_idx;
+
+	timeout_ms = 0;
+
+	if (timeout)
+		timeout_ms = static_cast<int>(timeout->tv_sec * 1000 + timeout->tv_usec / 1000);
+
+	ready = epoll_wait(fdw->epfd, fdw->ep_events, fdw->nfiles, timeout_ms);
+
+	if (ready == -1)
+		return -1;
+
+	event_idx = 0;
+
+	for (int i = 0; i < ready; ++i)
+	{
+		const int fd = fdw->ep_events[i].data.fd;
+		const uint32_t events = fdw->ep_events[i].events;
+
+		if (fd < 0 || fd >= fdw->nfiles)
+		{
+			sys_err("epoll ident overflow %d nfiles: %d", fd, fdw->nfiles);
+			continue;
+		}
+
+		if (events & (EPOLLERR | EPOLLHUP | EPOLLRDHUP))
+		{
+			fdw->ready_events[event_idx] = fdw->ep_events[i];
+			++event_idx;
+			continue;
+		}
+
+		if ((events & EPOLLIN) && event_idx < fdw->nfiles * 2)
+		{
+			fdw->ready_events[event_idx].data.fd = fd;
+			fdw->ready_events[event_idx].events = EPOLLIN;
+			++event_idx;
+		}
+
+		if ((events & EPOLLOUT) && event_idx < fdw->nfiles * 2)
+		{
+			fdw->ready_events[event_idx].data.fd = fd;
+			fdw->ready_events[event_idx].events = EPOLLOUT;
+			++event_idx;
+		}
+	}
+
+	return event_idx;
+}
+
+void* fdwatch_get_client_data(LPFDWATCH fdw, unsigned int event_idx)
+{
+	const int fd = fdw->ready_events[event_idx].data.fd;
+
+	assert(event_idx < fdw->nfiles * 2);
+
+	if (fd < 0 || fd >= fdw->nfiles)
+		return NULL;
+
+	return fdw->fd_data[fd];
+}
+
+int fdwatch_get_ident(LPFDWATCH fdw, unsigned int event_idx)
+{
+	assert(event_idx < fdw->nfiles * 2);
+	return fdw->ready_events[event_idx].data.fd;
+}
+
+void fdwatch_clear_event(LPFDWATCH fdw, socket_t fd, unsigned int event_idx)
+{
+	assert(event_idx < fdw->nfiles * 2);
+
+	if (fdw->ready_events[event_idx].data.fd != fd)
+		return;
+
+	fdw->ready_events[event_idx].data.fd = -1;
+	fdw->ready_events[event_idx].events = 0;
+}
+
+int fdwatch_check_event(LPFDWATCH fdw, socket_t fd, unsigned int event_idx)
+{
+	assert(event_idx < fdw->nfiles * 2);
+
+	if (fdw->ready_events[event_idx].data.fd != fd)
+		return 0;
+
+	if (fdw->ready_events[event_idx].events & (EPOLLERR | EPOLLHUP | EPOLLRDHUP))
+		return FDW_EOF;
+
+	if ((fdw->ready_events[event_idx].events & EPOLLIN) && (fdw->fd_rw[fd] & FDW_READ))
+		return FDW_READ;
+
+	if ((fdw->ready_events[event_idx].events & EPOLLOUT) && (fdw->fd_rw[fd] & FDW_WRITE))
+	{
+		if (fdw->fd_rw[fd] & FDW_WRITE_ONESHOT)
+		{
+			const int old_rw = fdw->fd_rw[fd];
+
+			fdw->fd_rw[fd] &= ~FDW_WRITE;
+
+			if (!fdwatch_update_registration(fdw, fd, old_rw))
+				sys_err("epoll_ctl write reset failed for fd %d [%d] %s", fd, errno, strerror(errno));
+		}
+
+		return FDW_WRITE;
+	}
+
+	return 0;
+}
+
+int fdwatch_get_buffer_size(LPFDWATCH fdw, socket_t fd)
+{
+	return INT_MAX;
+}
+
+#else

 #ifdef OS_WINDOWS
 static int win32_init_refcount = 0;
--- a/src/libthecore/fdwatch.h
+++ b/src/libthecore/fdwatch.h
@@ -1,73 +1,76 @@
 #pragma once

-#ifndef __USE_SELECT__
+typedef struct fdwatch		FDWATCH;
+typedef struct fdwatch *	LPFDWATCH;

-    typedef struct fdwatch		FDWATCH;
-    typedef struct fdwatch *	LPFDWATCH;
+enum EFdwatch
+{
+	FDW_NONE			= 0,
+	FDW_READ			= 1,
+	FDW_WRITE			= 2,
+	FDW_WRITE_ONESHOT	= 4,
+	FDW_EOF				= 8,
+};

-    enum EFdwatch
-    {
-		FDW_NONE			= 0,
-		FDW_READ			= 1,
-		FDW_WRITE			= 2,
-		FDW_WRITE_ONESHOT	= 4,
-		FDW_EOF				= 8,
-    };
+#if defined(OS_FREEBSD)

-    typedef struct kevent	KEVENT;
-    typedef struct kevent *	LPKEVENT;
-    typedef int				KQUEUE;
+typedef struct kevent	KEVENT;
+typedef struct kevent *	LPKEVENT;
+typedef int				KQUEUE;

-    struct fdwatch
-    {
-		KQUEUE		kq;
+struct fdwatch
+{
+	KQUEUE		kq;

-		int		nfiles;
+	int		nfiles;

-		LPKEVENT	kqevents;
-		int		nkqevents;
+	LPKEVENT	kqevents;
+	int		nkqevents;

-		LPKEVENT	kqrevents;
-		int *		fd_event_idx;
+	LPKEVENT	kqrevents;
+	int *		fd_event_idx;

-		void **		fd_data;
-		int *		fd_rw;
-    };
+	void **		fd_data;
+	int *		fd_rw;
+};
+
+#elif defined(__linux__)
+
+typedef struct epoll_event	EPOLL_EVENT;
+typedef struct epoll_event *	LPEPOLL_EVENT;
+
+struct fdwatch
+{
+	int				epfd;
+	int				nfiles;
+	LPEPOLL_EVENT	ep_events;
+	LPEPOLL_EVENT	ready_events;
+	void **			fd_data;
+	int *			fd_rw;
+};

 #else

-    typedef struct fdwatch		FDWATCH;
-    typedef struct fdwatch *	LPFDWATCH;
+struct fdwatch
+{
+	fd_set rfd_set;
+	fd_set wfd_set;

-    enum EFdwatch
-    {
-		FDW_NONE			= 0,
-		FDW_READ			= 1,
-		FDW_WRITE			= 2,
-		FDW_WRITE_ONESHOT	= 4,
-		FDW_EOF				= 8,
-    };
+	socket_t* select_fds;
+	int* select_rfdidx;

-    struct fdwatch
-    {
-		fd_set rfd_set;
-		fd_set wfd_set;
+	int nselect_fds;

-		socket_t* select_fds;
-		int* select_rfdidx;
+	fd_set working_rfd_set;
+	fd_set working_wfd_set;

-		int nselect_fds;
+	int nfiles;

-		fd_set working_rfd_set;
-		fd_set working_wfd_set;
+	void** fd_data;
+	int* fd_rw;
+};

-		int nfiles;
-
-		void** fd_data;
-		int* fd_rw;
-    };
-
-#endif // WIN32
+#endif


 LPFDWATCH	fdwatch_new(int nfiles);
@@ -82,4 +85,3 @@ void *		fdwatch_get_client_data(LPFDWATCH fdw, unsigned int event_idx);
 void		fdwatch_del_fd(LPFDWATCH fdw, socket_t fd);
 int			fdwatch_get_buffer_size(LPFDWATCH fdw, socket_t fd);
 int			fdwatch_get_ident(LPFDWATCH fdw, unsigned int event_idx);
-
--- a/src/libthecore/stdafx.h
+++ b/src/libthecore/stdafx.h
@@ -66,7 +66,7 @@ inline double rint(double x)

 #else

-#ifndef OS_FREEBSD
+#if !defined(OS_FREEBSD) && !defined(__linux__)
 #define __USE_SELECT__
 #ifdef __CYGWIN__
 #define _POSIX_SOURCE 1
@@ -102,6 +102,8 @@ inline double rint(double x)

 #ifdef OS_FREEBSD
 #include <sys/event.h>
+#elif defined(__linux__)
+#include <sys/epoll.h>
 #endif

 #endif