From: Egor <egor@patchysicp.xyz>
Date: Sat, 21 Oct 2023 15:25:30 +0000 (+0300)
Subject: add event loop abstraction
X-Git-Url: https://git.patchysicp.xyz/?a=commitdiff_plain;h=2143687391a1c2565eae856a84fbd8fdaca145b9;p=patchyircd.git

add event loop abstraction
---

diff --git a/Makefile b/Makefile
index 3cc9251..fe50f38 100644
--- a/Makefile
+++ b/Makefile
@@ -4,6 +4,7 @@ SOURCES := \
 	src/irc.c     \
 	src/ircfmt.c  \
 	src/net.c     \
+	src/netpoll.c \
 	src/netsock.c \
 	src/passwd.c  \
 	src/trie.c    \
@@ -14,6 +15,7 @@ HEADERS := \
 	src/irc.h     \
 	src/ircfmt.h  \
 	src/net.h     \
+	src/netpoll.h \
 	src/netsock.h \
 	src/passwd.h  \
 	src/trie.h    \
diff --git a/src/common.h b/src/common.h
index a9b55b2..b7209c6 100644
--- a/src/common.h
+++ b/src/common.h
@@ -1,6 +1,7 @@
 #ifndef COMMON_H_
 #define COMMON_H_
 #include "netsock.h"
+#include "netpoll.h"
 #include "config.h"
 #include "trie.h"
 #include <time.h>
@@ -60,6 +61,7 @@ struct client_data {
 	unsigned want_write_for_close :1;
 	unsigned dying :1;
 	unsigned write_error :1;
+	unsigned poll_reg :2;
 };
 struct channel {
 	char *key;
@@ -96,10 +98,5 @@ extern struct trie_node *channel_map;
 extern struct trie_node *autojoin;
 extern struct client_data  *clients;
 extern struct client_data  *dyingclients;
-extern        fd_set       readfds;
-extern struct sockaddr_in  serverAddrV4;
-extern struct sockaddr_in6 serverAddrV6;
-extern size_t              serverAddrV4len;
-extern size_t              serverAddrV6len;
 
 #endif
diff --git a/src/main.c b/src/main.c
index 739a6f0..3a36e1f 100644
--- a/src/main.c
+++ b/src/main.c
@@ -67,6 +67,8 @@ static void set_handler()
 		goto err;
 	if (net_nonblock(self_sock) == -1)
 		goto err;
+	net_poll_add(self_sock, SOCKTYPE_SELFPIPE, 0);
+	net_poll_reg(self_sock, NETPOLL_IN);
 	return;
 err:
 	net_perror("ERROR CREATING SELF SOCK");
@@ -77,8 +79,6 @@ err:
 
 // Client linked list
 struct client_data *clients = NULL, *dyingclients = NULL;
-// FDSET List
-fd_set readfds, writefds;
 
 FILE *fchanlog = NULL;
 
@@ -183,6 +183,8 @@ void accept_from_sock(struct listener *listener)
 		cl->peerport = port;
 		if (cctx)
 			cl->umode |= UMODE_SECURE;
+		net_poll_add(cl->FD, SOCKTYPE_CLIENT, cl);
+		update_poll_reg(cl);
 		extern const char *hostname;
 		copy_strf(&databuf, ":%s NOTICE * :Hello welcome to PatchyIRCD!\r\n", hostname);
 		sendbuf(cl);
@@ -200,6 +202,7 @@ int main(int argc, char *argv[])
 	fprintf(stderr, "[PatchyIRCD]: Starting up...\n");
 
 	net_init();
+	net_poll_init();
 	set_handler();
 	databuf.buf = malloc(sizeof(char) * 512);
 	databuf.cap = 512;
@@ -259,12 +262,16 @@ int main(int argc, char *argv[])
 #endif
 	}
 
-	for (int i = 0; i < listener_count; i++)
+	for (int i = 0; i < listener_count; i++) {
 		net_nonblock(listeners[i].sock);
+		net_poll_add(listeners[i].sock, SOCKTYPE_LISTENER, &listeners[i]);
+		net_poll_reg(listeners[i].sock, NETPOLL_IN);
+	}
 
 	FILE *f = fopen(getdatafile("init.txt"), "rb");
 	if (f) {
 		struct client_data *fake = allocClient();
+		fake->FD = INVALID_SOCKET;
 		fake->umode |= UMODE_OPER | UMODE_SECURE;
 		strcpy_lit(fake->buf, "NICK :root\r\n");
 		fake->bufLen = strlen(fake->buf);
@@ -320,8 +327,10 @@ int main(int argc, char *argv[])
 	while (is_running || dyingclients) {
 		if (is_siginted) {
 			fprintf(stderr, "[PatchyIRCD]: Cleaning up...\n");
-			for (int i = 0; i < listener_count; i++)
+			for (int i = 0; i < listener_count; i++) {
+				net_poll_del(listeners[i].sock);
 				closesocket(listeners[i].sock);
+			}
 			listener_count = 0;
 			for (struct client_data *cl = clients; cl;) {
 				struct client_data *next = cl->next;
@@ -333,95 +342,76 @@ int main(int argc, char *argv[])
 			if (!dyingclients)
 				break;
 		}
-		FD_ZERO(&readfds);
-		FD_ZERO(&writefds);
-		// Add listening SOCKETS to FDSET
-		for (int i = 0; i < listener_count; i++)
-			FD_SET(listeners[i].sock, &readfds);
-#ifdef _WIN32
-		if (self_sock != INVALID_SOCKET)
-			FD_SET(self_sock, &readfds);
-#endif
-		for (struct client_data *cl = clients; cl; cl = cl->next) {
-			if (!cl->want_write_for_read || cl->want_read_for_write)
-				FD_SET(cl->FD, &readfds);
-			if (cl->want_write_for_read || cl->want_write_for_write)
-				FD_SET(cl->FD, &writefds);
-		}
-		for (struct client_data *cl = dyingclients; cl; cl = cl->next) {
-			if (cl->want_read_for_write || cl->want_read_for_close)
-				FD_SET(cl->FD, &readfds);
-			if (cl->want_write_for_write || cl->want_write_for_close)
-				FD_SET(cl->FD, &writefds);
-		}
 
-		// Set highestFD
-#ifdef _WIN32
-		int highestFD = -1;
-#else
-		SOCKET highestFD = -1;
-		for (int i = 0; i < listener_count; i++)
-			if (highestFD < listeners[i].sock)
-				highestFD = listeners[i].sock;
-		for (struct client_data *cl = clients; cl; cl = cl->next)
-			if (highestFD < cl->FD)
-				highestFD = cl->FD;
-		for (struct client_data *cl = dyingclients; cl; cl = cl->next)
-			if (highestFD < cl->FD)
-				highestFD = cl->FD;
-#endif
-		// Wait until activity on sockets
-		if(select(highestFD+1,&readfds,&writefds,NULL,NULL) == -1) {
+		if (net_poll_begin() == -1) {
 #ifndef _WIN32
 			if (errno == EINTR)
 				continue;
 #endif
 			net_perror("ERROR SELECT");
 		}
+		int ev;
+		struct net_pollfd pfd;
+		while ((ev = net_poll_get(&pfd))) {
+			switch (pfd.type) {
 #ifdef _WIN32
-		if (self_sock != INVALID_SOCKET && FD_ISSET(self_sock, &readfds)) {
-			char x;
-			while (recv(self_sock, &x, 0, 0) != -1)
-				;
-		}
+			case SOCKTYPE_SELFPIPE:
+				if (ev & NETPOLL_IN) {
+					char x;
+					while (recv(pfd.fd, &x, 0, 0) != -1)
+						;
+				}
+				break;
 #endif
+			case SOCKTYPE_LISTENER:
+				if (ev & NETPOLL_IN)
+					accept_from_sock((struct listener *)pfd.assoc);
+				break;
+			case SOCKTYPE_CLIENT: {
+				struct client_data *cl = pfd.assoc;
+				int isr = ev & NETPOLL_IN;
+				int isw = ev & NETPOLL_OUT;
+				/* NOTE: recvProc can kill the client, so writing should be done first. */
+				if ((cl->want_read_for_write && isr) || (cl->want_write_for_write && isw))
+					sendClient(cl, 0, 0);
+				if (cl->want_write_for_read ? isw : isr)
+					recvProc(cl);
+				break;
+			}
+			case SOCKTYPE_DYING: {
+				struct client_data *cl = pfd.assoc;
+				int isr = ev & NETPOLL_IN;
+				int isw = ev & NETPOLL_OUT;
+				if (cl->wbufLen)
+					if ((cl->want_read_for_write && isr) || (cl->want_write_for_write && isw))
+						sendClient(cl, 0, 0);
+				if (!cl->wbufLen) {
+					if (!cl->want_read_for_close && !cl->want_write_for_close)
+						killConnectionPart2(cl);
+					else if ((cl->want_read_for_close && isr) || (cl->want_write_for_close && isw))
+						killConnectionPart2(cl);
+				}
+				break;
+			}
+			}
+		}
 
-		/* Accept new connections */
-		for (int i = 0; i < listener_count; i++)
-			if (FD_ISSET(listeners[i].sock, &readfds))
-				accept_from_sock(&listeners[i]);
-		/* Respond to already connected clients */
+		/* Clean up write error clients */
 		for (struct client_data *cl = clients; cl;) {
 			struct client_data *next = cl->next;
-			int isr = FD_ISSET(cl->FD, &readfds);
-			int isw = FD_ISSET(cl->FD, &writefds);
-			/* NOTE: recvProc can kill the client, so writing should be done first. */
-			if ((cl->want_read_for_write && isr) || (cl->want_write_for_write && isw))
-				sendClient(cl, 0, 0);
 			if (cl->write_error)
 				killConnection(cl, 0);
-			else if (cl->want_write_for_read ? isw : isr)
-				recvProc(cl);
-			cl = next;
-		}
-		/* Close dying clients */
-		for (struct client_data *cl = dyingclients; cl;) {
-			struct client_data *next = cl->next;
-			int isr = FD_ISSET(cl->FD, &readfds);
-			int isw = FD_ISSET(cl->FD, &writefds);
-			if (cl->wbufLen)
-				if ((cl->want_read_for_write && isr) || (cl->want_write_for_write && isw))
-					sendClient(cl, 0, 0);
-			if (!cl->wbufLen) {
-				if (!cl->want_read_for_close && !cl->want_write_for_close)
-					killConnectionPart2(cl);
-				else if ((cl->want_read_for_close && isr) || (cl->want_write_for_close && isw))
-					killConnectionPart2(cl);
-			}
 			cl = next;
 		}
 	}
 
+#ifdef _WIN32
+	if (self_sock != INVALID_SOCKET) {
+		net_poll_del(self_sock);
+		closesocket(self_sock);
+	}
+#endif
+	net_poll_fini();
 	net_cleanup();
 #ifdef HAS_TLS
 	tls_free(tls_ctx);
diff --git a/src/net.c b/src/net.c
index 968935b..347dfe6 100644
--- a/src/net.c
+++ b/src/net.c
@@ -7,13 +7,39 @@ struct client_data *allocClient()
 	return calloc(1, sizeof(struct client_data));
 }
 
+void update_poll_reg(struct client_data *cl)
+{
+	if (cl->FD == INVALID_SOCKET)
+		return;
+	int nreg = 0;
+	/* Shorthand:
+	 * r = (!d && !w4r) || r4w || (d && r4c)
+	 * w = (!d &&  w4r) || w4w || (d && w4c)
+	 */
+	if (!cl->dying) {
+		if (!cl->want_write_for_read || cl->want_read_for_write)
+			nreg |= NETPOLL_IN;
+		if (cl->want_write_for_read || cl->want_write_for_write)
+			nreg |= NETPOLL_OUT;
+	} else {
+		if (cl->want_read_for_write || cl->want_read_for_close)
+			nreg |= NETPOLL_IN;
+		if (cl->want_write_for_write || cl->want_write_for_close)
+			nreg |= NETPOLL_OUT;
+	}
+	if (nreg != cl->poll_reg) {
+		net_poll_reg(cl->FD, nreg);
+		cl->poll_reg = nreg;
+	}
+}
+
 void killConnection(struct client_data *cl, int byChoice)
 {
 	if (!cl)
 		return;
 	if (cl->dying)
 		return;
-	cl->dying = 1; /* NOTE: this recursion guard isn't necessary, but it's there just in case */
+	cl->dying = 1;
 	if (cl->peername)
 		fprintf(stderr, "Client disconnected from Socket: %"PRISocket", IP: %s PORT: %d\n", cl->FD, cl->peername, cl->peerport);
 	if (!byChoice && cl->nick && cl->username) {
@@ -32,6 +58,11 @@ void killConnection(struct client_data *cl, int byChoice)
 	}
 	cl->next = dyingclients;
 	dyingclients = cl;
+	if (cl->FD != INVALID_SOCKET) {
+		net_poll_del(cl->FD);
+		net_poll_add(cl->FD, SOCKTYPE_DYING, cl);
+	}
+	update_poll_reg(cl);
 	/* Part all channels */
 	struct trie_find *f = NULL;
 	if (trie_findfirst(&f, &cl->channels) == 1)
@@ -66,10 +97,12 @@ void killConnectionPart2(struct client_data *cl)
 			int rv = tls_close(cl->cctx);
 			if (rv == TLS_WANT_POLLIN) {
 				cl->want_read_for_close = 1;
+				update_poll_reg(cl);
 				return;
 			}
 			if (rv == TLS_WANT_POLLOUT) {
 				cl->want_write_for_close = 1;
+				update_poll_reg(cl);
 				return;
 			}
 			if (rv == -1) {
@@ -80,6 +113,7 @@ void killConnectionPart2(struct client_data *cl)
 		}
 #endif
 		shutdown(cl->FD, SHUT_RDWR);
+		net_poll_del(cl->FD);
 		closesocket(cl->FD);
 	}
 	/* Remove the client from the dyingclients chain */
@@ -127,9 +161,11 @@ void recvProc(struct client_data *cl) {
 					killConnection(cl, 0);
 					return;
 				case TLS_WANT_POLLIN:
+					update_poll_reg(cl);
 					return;
 				case TLS_WANT_POLLOUT:
 					cl->want_write_for_read = 1;
+					update_poll_reg(cl);
 					return;
 			}
 		}
@@ -147,6 +183,8 @@ void recvProc(struct client_data *cl) {
 					if(!net_wouldblock()){
 						net_perror("RECV");
 						killConnection(cl, 0);
+					} else {
+						update_poll_reg(cl);
 					}
 					return;
 			}
@@ -157,8 +195,12 @@ void recvProc(struct client_data *cl) {
 			if(!x) break;
 			packetReturn = parsePacket(cl);
 			// Client QUIT
-			if (packetReturn == 1 || cl->write_error)
+			if (packetReturn == 1)
 				return;
+			if (cl->write_error) {
+				update_poll_reg(cl);
+				return;
+			}
 			memmove(cl->buf, x+1, cl->buf + cl->bufLen-(x+1));
 			cl->bufLen -= x+1 - cl->buf;
 		}
@@ -225,6 +267,7 @@ void sendClient(struct client_data *cl, const char *buf, size_t len)
 	if (!cl->wbufLen) {
 		if (len) {
 			int left = sendClientRaw(cl, buf, len);
+			update_poll_reg(cl);
 			if (left == -1) {
 				/* Write error happened, discard the buffer and close the
 				 * connection.  sendClient is used from a ton of places,
@@ -248,6 +291,7 @@ void sendClient(struct client_data *cl, const char *buf, size_t len)
 		cl->wbufLen += len;
 	}
 	int left = sendClientRaw(cl, cl->wbuf, cl->wbufLen);
+	update_poll_reg(cl);
 	if (left == -1) {
 		cl->wbufLen = 0;
 		cl->write_error = 1;
diff --git a/src/net.h b/src/net.h
index d0d4bde..d339065 100644
--- a/src/net.h
+++ b/src/net.h
@@ -5,6 +5,7 @@
 void sendbuf(struct client_data *client);
 void sendbuf_visible(struct client_data *client);
 struct client_data *allocClient();
+void update_poll_reg(struct client_data *cl);
 void killConnection(struct client_data *client, int byChoice);
 void killConnectionPart2(struct client_data *cl);
 void recvProc(struct client_data *cl);
diff --git a/src/netpoll.c b/src/netpoll.c
new file mode 100644
index 0000000..88355d0
--- /dev/null
+++ b/src/netpoll.c
@@ -0,0 +1,137 @@
+#include "netpoll.h"
+#define NETPOLL_MAXFDS 1024
+static struct net_pollfd fds[NETPOLL_MAXFDS] = {0};
+static int nfds = 0;
+
+static int search(SOCKET fd) {
+	int l = 0, r = nfds;
+	while (l != r) {
+		int m = l + (r-l)/2;
+		if (fds[m].fd < fd)
+			l = m+1;
+		else if (fd < fds[m].fd)
+			r = m;
+		else
+			l = r = m;
+	}
+	return l;
+}
+
+#if defined(NETPOLL_USE_SELECT)
+static fd_set readfds, writefds, xreadfds, xwritefds;
+static int lastptr = 0;
+void net_poll_init(void)
+{
+	FD_ZERO(&readfds);
+	FD_ZERO(&writefds);
+}
+void net_poll_fini(void)
+{
+}
+void net_poll_add_impl(int i)
+{
+	if (fds[i].fd >= FD_SETSIZE) {
+		fprintf(stderr, "net_poll_add_impl: FD_SETSIZE limit\n");
+		abort();
+	}
+	if (lastptr >= i)
+		lastptr++;
+}
+void net_poll_del_impl(int i)
+{
+	if (lastptr > i)
+		lastptr--;
+}
+void net_poll_reg_impl(int i, int ev)
+{
+	if (ev & NETPOLL_IN)
+		FD_SET(fds[i].fd, &readfds);
+	else
+		FD_CLR(fds[i].fd, &readfds);
+	if (ev & NETPOLL_OUT)
+		FD_SET(fds[i].fd, &writefds);
+	else
+		FD_CLR(fds[i].fd, &writefds);
+}
+int net_poll_begin(void)
+{
+	memcpy(&xreadfds, &readfds, sizeof(readfds));
+	memcpy(&xwritefds, &writefds, sizeof(writefds));
+#ifdef _WIN32
+	int highfd = 0;
+#else
+	int highfd = nfds ? fds[nfds-1].fd+1 : 0;
+#endif
+	int rv = select(highfd, &xreadfds, &xwritefds, 0, 0);
+	lastptr = 0;
+	return rv;
+}
+int net_poll_get(struct net_pollfd *out)
+{
+	int ev = 0;
+	while (!ev) {
+		if (lastptr == nfds)
+			return 0;
+		if (FD_ISSET(fds[lastptr].fd, &xreadfds))
+			ev |= NETPOLL_IN;
+		if (FD_ISSET(fds[lastptr].fd, &xwritefds))
+			ev |= NETPOLL_OUT;
+		lastptr++;
+	}
+	*out = fds[lastptr-1];
+	return ev;
+}
+#else
+#error pick a netpoll implementation
+#endif
+void net_poll_add(SOCKET fd, int type, void *assoc)
+{
+	if (fd == INVALID_SOCKET) {
+		fprintf(stderr, "net_poll_add: INVALID_SOCKET\n");
+		abort();
+	}
+	int i = search(fd);
+	if (i < nfds && fds[i].fd == fd) {
+		fprintf(stderr, "net_poll_add: socket already added\n");
+		abort();
+	}
+	if (i == NETPOLL_MAXFDS) {
+		fprintf(stderr, "net_poll_add: NETPOLL_MAXFDS limit\n");
+		abort();
+	}
+	memmove(fds+i+1, fds+i, sizeof(fds[0])*(nfds-i));
+	nfds++;
+	fds[i].fd = fd;
+	fds[i].type = type;
+	fds[i].assoc = assoc;
+	net_poll_add_impl(i);
+}
+void net_poll_del(SOCKET fd)
+{
+	if (fd == INVALID_SOCKET) {
+		fprintf(stderr, "net_poll_del: INVALID_SOCKET\n");
+		abort();
+	}
+	int i = search(fd);
+	if (i >= nfds || fds[i].fd != fd) {
+		fprintf(stderr, "net_poll_del: socket not added\n");
+		abort();
+	}
+	net_poll_reg_impl(i, 0);
+	net_poll_del_impl(i);
+	memmove(fds+i, fds+i+1, sizeof(fds[0])*(nfds-(i+1)));
+	nfds--;
+}
+void net_poll_reg(SOCKET fd, int ev)
+{
+	if (fd == INVALID_SOCKET) {
+		fprintf(stderr, "net_poll_reg: INVALID_SOCKET\n");
+		abort();
+	}
+	int i = search(fd);
+	if (i >= nfds || fds[i].fd != fd) {
+		fprintf(stderr, "net_poll_reg: socket not added\n");
+		abort();
+	}
+	net_poll_reg_impl(i, ev);
+}
diff --git a/src/netpoll.h b/src/netpoll.h
new file mode 100644
index 0000000..d4f1f43
--- /dev/null
+++ b/src/netpoll.h
@@ -0,0 +1,26 @@
+#ifndef NETPOLL_H_
+#define NETPOLL_H_
+#include "netsock.h"
+#define NETPOLL_USE_SELECT
+struct net_pollfd {
+	void *assoc;
+	SOCKET fd;
+	int type;
+};
+void net_poll_init(void);
+void net_poll_fini(void);
+void net_poll_add(SOCKET fd, int type, void *assoc);
+void net_poll_del(SOCKET fd);
+void net_poll_reg(SOCKET fd, int ev);
+int net_poll_begin(void);
+int net_poll_get(struct net_pollfd *out);
+#define NETPOLL_IN 1
+#define NETPOLL_OUT 2
+
+#define SOCKTYPE_LISTENER 0
+#define SOCKTYPE_SELFPIPE 1
+#define SOCKTYPE_CLIENT 2
+#define SOCKTYPE_DYING 3
+#define SOCKTYPE_MAX 4
+
+#endif