15 files changed, 766 insertions, 230 deletions

diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt
index af6e9fb..fb94b3a 100644
--- a/src/CMakeLists.txt
+++ b/src/CMakeLists.txt
@@ -39,6 +39,7 @@ generate_sql_header(sqlite)
 add_my_executable(server server_main.c server.c
     cmd_line.c
     command.c
+    event_loop.c
     file.c
     msg.c
     net.c
@@ -48,6 +49,7 @@ add_my_executable(server server_main.c server.c
     sqlite.c
     storage.c
     storage_sqlite.c
+    string.c
     tcp_server.c
     worker_queue.c)
 target_link_libraries(server PRIVATE pthread sqlite3)
@@ -63,11 +65,13 @@ add_my_executable(worker worker_main.c worker.c
     ci.c
     cmd_line.c
     command.c
+    event_loop.c
     file.c
     git.c
     msg.c
     net.c
     process.c
     run_queue.c
-    signal.c)
+    signal.c
+    string.c)
 target_link_libraries(worker PRIVATE git2 pthread)
diff --git a/src/command.c b/src/command.c
index bd58f05..6306f7e 100644
--- a/src/command.c
+++ b/src/command.c
@@ -6,6 +6,8 @@
  */
 
 #include "command.h"
+#include "compiler.h"
+#include "event_loop.h"
 #include "log.h"
 #include "msg.h"
 
@@ -128,20 +130,29 @@ int cmd_dispatcher_handle(const struct cmd_dispatcher *dispatcher, const struct
 	return cmd_dispatcher_handle_internal(dispatcher, command, result, dispatcher->ctx);
 }
 
+static struct cmd_conn_ctx *make_conn_ctx(int fd, void *arg)
+{
+	struct cmd_conn_ctx *ctx = malloc(sizeof(struct cmd_conn_ctx));
+	if (!ctx) {
+		log_errno("malloc");
+		return NULL;
+	}
+
+	ctx->fd = fd;
+	ctx->arg = arg;
+
+	return ctx;
+}
+
 int cmd_dispatcher_handle_conn(int conn_fd, void *_dispatcher)
 {
 	struct cmd_dispatcher *dispatcher = (struct cmd_dispatcher *)_dispatcher;
 	struct msg *request = NULL, *response = NULL;
 	int ret = 0;
 
-	struct cmd_conn_ctx *new_ctx = malloc(sizeof(struct cmd_conn_ctx));
-	if (!new_ctx) {
-		log_errno("malloc");
+	struct cmd_conn_ctx *new_ctx = make_conn_ctx(conn_fd, dispatcher->ctx);
+	if (!new_ctx)
 		return -1;
-	}
-
-	new_ctx->fd = conn_fd;
-	new_ctx->arg = dispatcher->ctx;
 
 	ret = msg_recv(conn_fd, &request);
 	if (ret < 0)
@@ -168,3 +179,57 @@ free_ctx:
 
 	return ret;
 }
+
+static int cmd_dispatcher_handle_event(UNUSED struct event_loop *loop, int fd, short revents,
+                                       void *_dispatcher)
+{
+	struct cmd_dispatcher *dispatcher = (struct cmd_dispatcher *)_dispatcher;
+	struct msg *request = NULL, *response = NULL;
+	int ret = 0;
+
+	if (!(revents & POLLIN)) {
+		log_err("Descriptor %d is not readable\n", fd);
+		return -1;
+	}
+
+	struct cmd_conn_ctx *new_ctx = make_conn_ctx(fd, dispatcher->ctx);
+	if (!new_ctx)
+		return -1;
+
+	ret = msg_recv(fd, &request);
+	if (ret < 0)
+		goto free_ctx;
+
+	ret = cmd_dispatcher_handle_internal(dispatcher, request, &response, new_ctx);
+	if (ret < 0)
+		goto free_response;
+
+	if (response) {
+		ret = msg_send(fd, response);
+		if (ret < 0)
+			goto free_response;
+	}
+
+free_response:
+	if (response)
+		msg_free(response);
+
+	msg_free(request);
+
+free_ctx:
+	free(new_ctx);
+
+	return ret;
+}
+
+int cmd_dispatcher_add_to_event_loop(const struct cmd_dispatcher *dispatcher,
+                                     struct event_loop *loop, int fd)
+{
+	struct event_fd entry = {
+	    .fd = fd,
+	    .events = POLLIN,
+	    .handler = cmd_dispatcher_handle_event,
+	    .arg = (void *)dispatcher,
+	};
+	return event_loop_add(loop, &entry);
+}
diff --git a/src/command.h b/src/command.h
index 90facbb..3d13abd 100644
--- a/src/command.h
+++ b/src/command.h
@@ -8,6 +8,7 @@
 #ifndef __COMMAND_H__
 #define __COMMAND_H__
 
+#include "event_loop.h"
 #include "msg.h"
 
 #include <stddef.h>
@@ -28,6 +29,8 @@ void cmd_dispatcher_destroy(struct cmd_dispatcher *);
 int cmd_dispatcher_handle(const struct cmd_dispatcher *, const struct msg *command,
                           struct msg **response);
 
+int cmd_dispatcher_add_to_event_loop(const struct cmd_dispatcher *, struct event_loop *, int fd);
+
 struct cmd_conn_ctx {
 	int fd;
 	void *arg;
diff --git a/src/event_loop.c b/src/event_loop.c
new file mode 100644
index 0000000..8d91480
--- /dev/null
+++ b/src/event_loop.c
@@ -0,0 +1,204 @@
+/*
+ * Copyright (c) 2023 Egor Tensin <Egor.Tensin@gmail.com>
+ * This file is part of the "cimple" project.
+ * For details, see https://github.com/egor-tensin/cimple.
+ * Distributed under the MIT License.
+ */
+
+#include "event_loop.h"
+#include "log.h"
+#include "net.h"
+#include "string.h"
+
+#include <poll.h>
+#include <stddef.h>
+#include <stdlib.h>
+#include <sys/queue.h>
+
+SIMPLEQ_HEAD(event_fd_queue, event_fd);
+
+static struct event_fd *event_fd_copy(const struct event_fd *src)
+{
+	struct event_fd *res = malloc(sizeof(*src));
+	if (!res) {
+		log_errno("malloc");
+		return NULL;
+	}
+	*res = *src;
+	return res;
+}
+
+static void event_fd_destroy(struct event_fd *entry)
+{
+	free(entry);
+}
+
+static void event_fd_queue_create(struct event_fd_queue *queue)
+{
+	SIMPLEQ_INIT(queue);
+}
+
+static void event_fd_queue_destroy(struct event_fd_queue *queue)
+{
+	struct event_fd *entry1 = SIMPLEQ_FIRST(queue);
+	while (entry1) {
+		struct event_fd *entry2 = SIMPLEQ_NEXT(entry1, entries);
+		event_fd_destroy(entry1);
+		entry1 = entry2;
+	}
+	SIMPLEQ_INIT(queue);
+}
+
+struct event_loop {
+	nfds_t nfds;
+	struct event_fd_queue entries;
+};
+
+int event_loop_create(struct event_loop **_loop)
+{
+	int ret = 0;
+
+	struct event_loop *loop = calloc(1, sizeof(struct event_loop));
+	if (!loop) {
+		log_errno("calloc");
+		return -1;
+	}
+	*_loop = loop;
+
+	event_fd_queue_create(&loop->entries);
+
+	return ret;
+}
+
+void event_loop_destroy(struct event_loop *loop)
+{
+	event_fd_queue_destroy(&loop->entries);
+	free(loop);
+}
+
+int event_loop_add(struct event_loop *loop, const struct event_fd *entry)
+{
+	log("Adding descriptor %d to event loop\n", entry->fd);
+
+	struct event_fd *copied = event_fd_copy(entry);
+	if (!copied)
+		return -1;
+
+	nfds_t nfds = loop->nfds + 1;
+	SIMPLEQ_INSERT_TAIL(&loop->entries, copied, entries);
+	loop->nfds = nfds;
+
+	return 0;
+}
+
+static void event_loop_remove(struct event_loop *loop, struct event_fd *entry)
+{
+	log("Removing descriptor %d from event loop\n", entry->fd);
+
+	SIMPLEQ_REMOVE(&loop->entries, entry, event_fd, entries);
+	net_close(entry->fd);
+	event_fd_destroy(entry);
+	--loop->nfds;
+}
+
+static char *append_event(char *buf, size_t sz, char *ptr, const char *event)
+{
+	if (ptr > buf)
+		ptr = stpecpy(ptr, buf + sz, ",");
+	return stpecpy(ptr, buf + sz, event);
+}
+
+static char *events_to_string(short events)
+{
+	const size_t sz = 128;
+	char *buf = calloc(1, sz);
+	if (!buf)
+		return NULL;
+
+	char *ptr = buf;
+
+	if (events & POLLNVAL)
+		ptr = append_event(buf, sz, ptr, "POLLNVAL");
+	if (events & POLLERR)
+		ptr = append_event(buf, sz, ptr, "POLLERR");
+	if (events & POLLHUP)
+		ptr = append_event(buf, sz, ptr, "POLLHUP");
+	if (events & POLLIN)
+		ptr = append_event(buf, sz, ptr, "POLLIN");
+	if (events & POLLOUT)
+		ptr = append_event(buf, sz, ptr, "POLLOUT");
+
+	return buf;
+}
+
+static struct pollfd *make_pollfds(const struct event_loop *loop)
+{
+	struct pollfd *fds = calloc(loop->nfds, sizeof(struct pollfd));
+	if (!fds) {
+		log_errno("calloc");
+		return NULL;
+	}
+
+	struct event_fd *entry = SIMPLEQ_FIRST(&loop->entries);
+	for (nfds_t i = 0; i < loop->nfds; ++i, entry = SIMPLEQ_NEXT(entry, entries)) {
+		fds[i].fd = entry->fd;
+		fds[i].events = entry->events;
+	}
+
+	log("Descriptors:\n");
+	for (nfds_t i = 0; i < loop->nfds; ++i) {
+		char *events = events_to_string(fds[i].events);
+		log("    %d (%s)\n", fds[i].fd, events ? events : "");
+		free(events);
+	}
+
+	return fds;
+}
+
+int event_loop_run(struct event_loop *loop)
+{
+	struct pollfd *fds = make_pollfds(loop);
+	if (!fds)
+		return -1;
+
+	int ret = poll(fds, loop->nfds, -1);
+	if (ret < 0) {
+		log_errno("poll");
+		return ret;
+	}
+	ret = 0;
+
+	struct event_fd *entry = SIMPLEQ_FIRST(&loop->entries);
+	for (nfds_t i = 0; i < loop->nfds; ++i) {
+		struct event_fd *next = SIMPLEQ_NEXT(entry, entries);
+
+		if (!fds[i].revents)
+			goto next;
+
+		char *events = events_to_string(fds[i].revents);
+		log("Descriptor %d is ready: %s\n", fds[i].fd, events ? events : "");
+		free(events);
+
+		/* Execute all handlers but notice if any of them fail. */
+		const int handler_ret = entry->handler(loop, fds[i].fd, fds[i].revents, entry->arg);
+		switch (handler_ret) {
+		case 0:
+			goto next;
+		case EVENT_LOOP_REMOVE:
+			goto remove;
+		default:
+			break;
+		}
+
+	remove:
+		event_loop_remove(loop, entry);
+		goto next;
+
+	next:
+		entry = next;
+		continue;
+	}
+
+	free(fds);
+	return ret;
+}
diff --git a/src/event_loop.h b/src/event_loop.h
new file mode 100644
index 0000000..eb5ff05
--- /dev/null
+++ b/src/event_loop.h
@@ -0,0 +1,36 @@
+/*
+ * Copyright (c) 2023 Egor Tensin <Egor.Tensin@gmail.com>
+ * This file is part of the "cimple" project.
+ * For details, see https://github.com/egor-tensin/cimple.
+ * Distributed under the MIT License.
+ */
+
+#ifndef __EVENT_LOOP_H__
+#define __EVENT_LOOP_H__
+
+#include <poll.h>
+#include <sys/queue.h>
+
+struct event_loop;
+
+int event_loop_create(struct event_loop **);
+void event_loop_destroy(struct event_loop *);
+
+int event_loop_run(struct event_loop *);
+
+#define EVENT_LOOP_REMOVE 1
+
+typedef int (*event_loop_handler)(struct event_loop *, int fd, short revents, void *arg);
+
+struct event_fd {
+	int fd;
+	short events;
+	event_loop_handler handler;
+	void *arg;
+
+	SIMPLEQ_ENTRY(event_fd) entries;
+};
+
+int event_loop_add(struct event_loop *, const struct event_fd *);
+
+#endif
diff --git a/src/server.c b/src/server.c
index a5319e0..2839a61 100644
--- a/src/server.c
+++ b/src/server.c
@@ -9,6 +9,7 @@
 #include "command.h"
 #include "compiler.h"
 #include "const.h"
+#include "event_loop.h"
 #include "log.h"
 #include "msg.h"
 #include "run_queue.h"
@@ -27,6 +28,11 @@ struct server {
 
 	int stopping;
 
+	struct cmd_dispatcher *cmd_dispatcher;
+
+	struct event_loop *event_loop;
+	int signalfd;
+
 	struct worker_queue worker_queue;
 	struct run_queue run_queue;
 
@@ -67,8 +73,10 @@ static void server_notify(struct server *server)
 	pthread_errno_if(pthread_cond_signal(&server->server_cv), "pthread_cond_signal");
 }
 
-static int server_set_stopping(struct server *server)
+static int server_set_stopping(UNUSED struct event_loop *loop, UNUSED int fd, UNUSED short revents,
+                               void *_server)
 {
+	struct server *server = (struct server *)_server;
 	int ret = 0;
 
 	ret = server_lock(server);
@@ -117,7 +125,7 @@ static int server_enqueue_run(struct server *server, struct run *run)
 		return ret;
 
 	run_queue_add_last(&server->run_queue, run);
-	log("Added a new CI run for repository %s to the queue\n", run_get_url(run));
+	log("Added a new run for repository %s to the queue\n", run_get_url(run));
 
 	server_notify(server);
 	server_unlock(server);
@@ -147,25 +155,26 @@ static int server_assign_run(struct server *server)
 	int ret = 0;
 
 	struct run *run = run_queue_remove_first(&server->run_queue);
-	log("Removed a CI run for repository %s from the queue\n", run_get_url(run));
+	log("Removed run for repository %s from the queue\n", run_get_url(run));
 
 	struct worker *worker = worker_queue_remove_first(&server->worker_queue);
 	log("Removed worker %d from the queue\n", worker_get_fd(worker));
 
 	const char *argv[] = {CMD_RUN, run_get_url(run), run_get_rev(run), NULL};
-
 	ret = msg_talk_argv(worker_get_fd(worker), argv, NULL);
+
 	if (ret < 0) {
-		/* Failed to communicate with the worker, requeue the run
-		 * and forget about the worker. */
-		worker_destroy(worker);
+		log("Failed to assign run for repository %s to worker %d, requeueing\n",
+		    run_get_url(run), worker_get_fd(worker));
 		run_queue_add_first(&server->run_queue, run);
-		return ret;
+	} else {
+		log("Assigned run for repository %s to worker %d\n", run_get_url(run),
+		    worker_get_fd(worker));
+		run_destroy(run);
 	}
 
-	/* Send the run to the worker, forget about both of them for a while. */
 	worker_destroy(worker);
-	run_destroy(run);
+
 	return ret;
 }
 
@@ -198,15 +207,89 @@ exit:
 	return NULL;
 }
 
-int server_create(struct server **_server, const struct settings *settings)
+static int handle_cmd_new_worker(UNUSED const struct msg *request, UNUSED struct msg **response,
+                                 void *_ctx)
 {
-	struct storage_settings storage_settings;
+	struct cmd_conn_ctx *ctx = (struct cmd_conn_ctx *)_ctx;
+	struct server *server = (struct server *)ctx->arg;
+	int client_fd = ctx->fd;
+
+	struct worker *worker = NULL;
+	int ret = 0;
+
+	ret = worker_create(&worker, client_fd);
+	if (ret < 0)
+		return ret;
+
+	ret = server_enqueue_worker(server, worker);
+	if (ret < 0)
+		goto destroy_worker;
+
+	return ret;
+
+destroy_worker:
+	worker_destroy(worker);
+
+	return ret;
+}
+
+static int handle_cmd_run(const struct msg *request, struct msg **response, void *_ctx)
+{
+	struct cmd_conn_ctx *ctx = (struct cmd_conn_ctx *)_ctx;
+	struct server *server = (struct server *)ctx->arg;
+	struct run *run = NULL;
+
 	int ret = 0;
 
-	ret = signal_handle_stops();
+	ret = run_from_msg(&run, request);
 	if (ret < 0)
 		return ret;
 
+	ret = msg_success(response);
+	if (ret < 0)
+		goto destroy_run;
+
+	ret = server_enqueue_run(server, run);
+	if (ret < 0)
+		goto free_response;
+
+	return ret;
+
+free_response:
+	msg_free(*response);
+	*response = NULL;
+
+destroy_run:
+	run_destroy(run);
+
+	return ret;
+}
+
+static int handle_cmd_complete(UNUSED const struct msg *request, UNUSED struct msg **response,
+                               void *_ctx)
+{
+	struct cmd_conn_ctx *ctx = (struct cmd_conn_ctx *)_ctx;
+	int client_fd = ctx->fd;
+	int ret = 0;
+
+	log("Received a \"run complete\" message from worker %d\n", client_fd);
+
+	return ret;
+}
+
+static struct cmd_desc commands[] = {
+    {CMD_NEW_WORKER, handle_cmd_new_worker},
+    {CMD_RUN, handle_cmd_run},
+    {CMD_COMPLETE, handle_cmd_complete},
+};
+
+static const size_t numof_commands = sizeof(commands) / sizeof(commands[0]);
+
+int server_create(struct server **_server, const struct settings *settings)
+{
+	struct storage_settings storage_settings;
+	int ret = 0;
+
 	struct server *server = malloc(sizeof(struct server));
 	if (!server) {
 		log_errno("malloc");
@@ -227,6 +310,24 @@ int server_create(struct server **_server, const struct settings *settings)
 
 	server->stopping = 0;
 
+	ret = cmd_dispatcher_create(&server->cmd_dispatcher, commands, numof_commands, server);
+	if (ret < 0)
+		goto destroy_cv;
+
+	ret = event_loop_create(&server->event_loop);
+	if (ret < 0)
+		goto destroy_cmd_dispatcher;
+
+	ret = signalfd_listen_for_stops();
+	if (ret < 0)
+		goto destroy_event_loop;
+	server->signalfd = ret;
+
+	ret = signalfd_add_to_event_loop(server->signalfd, server->event_loop, server_set_stopping,
+	                                 server);
+	if (ret < 0)
+		goto close_signalfd;
+
 	worker_queue_create(&server->worker_queue);
 	run_queue_create(&server->run_queue);
 
@@ -239,10 +340,15 @@ int server_create(struct server **_server, const struct settings *settings)
 	if (ret < 0)
 		goto destroy_run_queue;
 
-	ret = tcp_server_create(&server->tcp_server, settings->port);
+	ret = tcp_server_create(&server->tcp_server, settings->port, cmd_dispatcher_handle_conn,
+	                        server->cmd_dispatcher);
 	if (ret < 0)
 		goto destroy_storage;
 
+	ret = tcp_server_add_to_event_loop(server->tcp_server, server->event_loop);
+	if (ret < 0)
+		goto destroy_tcp_server;
+
 	ret = pthread_create(&server->main_thread, NULL, server_main_thread, server);
 	if (ret) {
 		pthread_errno(ret, "pthread_create");
@@ -263,6 +369,16 @@ destroy_run_queue:
 
 	worker_queue_destroy(&server->worker_queue);
 
+close_signalfd:
+	signalfd_destroy(server->signalfd);
+
+destroy_event_loop:
+	event_loop_destroy(server->event_loop);
+
+destroy_cmd_dispatcher:
+	cmd_dispatcher_destroy(server->cmd_dispatcher);
+
+destroy_cv:
 	pthread_errno_if(pthread_cond_destroy(&server->server_cv), "pthread_cond_destroy");
 
 destroy_mtx:
@@ -283,128 +399,27 @@ void server_destroy(struct server *server)
 	storage_destroy(&server->storage);
 	run_queue_destroy(&server->run_queue);
 	worker_queue_destroy(&server->worker_queue);
+	signalfd_destroy(server->signalfd);
+	event_loop_destroy(server->event_loop);
+	cmd_dispatcher_destroy(server->cmd_dispatcher);
 	pthread_errno_if(pthread_cond_destroy(&server->server_cv), "pthread_cond_destroy");
 	pthread_errno_if(pthread_mutex_destroy(&server->server_mtx), "pthread_mutex_destroy");
 	free(server);
 }
 
-static int handle_cmd_new_worker(UNUSED const struct msg *request, UNUSED struct msg **response,
-                                 void *_ctx)
-{
-	struct cmd_conn_ctx *ctx = (struct cmd_conn_ctx *)_ctx;
-	struct server *server = (struct server *)ctx->arg;
-	int client_fd = ctx->fd;
-
-	struct worker *worker = NULL;
-	int ret = 0;
-
-	ret = worker_create(&worker, client_fd);
-	if (ret < 0)
-		return ret;
-
-	ret = server_enqueue_worker(server, worker);
-	if (ret < 0)
-		goto destroy_worker;
-
-	return ret;
-
-destroy_worker:
-	worker_destroy(worker);
-
-	return ret;
-}
-
-static int handle_cmd_run(const struct msg *request, struct msg **response, void *_ctx)
-{
-	struct cmd_conn_ctx *ctx = (struct cmd_conn_ctx *)_ctx;
-	struct server *server = (struct server *)ctx->arg;
-	struct run *run = NULL;
-
-	int ret = 0;
-
-	ret = run_from_msg(&run, request);
-	if (ret < 0)
-		return ret;
-
-	ret = msg_success(response);
-	if (ret < 0)
-		goto destroy_run;
-
-	ret = server_enqueue_run(server, run);
-	if (ret < 0)
-		goto free_response;
-
-	return ret;
-
-free_response:
-	msg_free(*response);
-
-destroy_run:
-	run_destroy(run);
-
-	return ret;
-}
-
-static int handle_cmd_complete(UNUSED const struct msg *request, UNUSED struct msg **response,
-                               void *_ctx)
-{
-	struct cmd_conn_ctx *ctx = (struct cmd_conn_ctx *)_ctx;
-	struct server *server = (struct server *)ctx->arg;
-	int client_fd = ctx->fd;
-
-	struct worker *worker = NULL;
-	int ret = 0;
-
-	log("Received a \"run complete\" message from worker %d\n", client_fd);
-
-	ret = worker_create(&worker, client_fd);
-	if (ret < 0)
-		return ret;
-
-	ret = server_enqueue_worker(server, worker);
-	if (ret < 0)
-		goto destroy_worker;
-
-	return ret;
-
-destroy_worker:
-	worker_destroy(worker);
-
-	return 0;
-}
-
-static struct cmd_desc commands[] = {
-    {CMD_NEW_WORKER, handle_cmd_new_worker},
-    {CMD_RUN, handle_cmd_run},
-    {CMD_COMPLETE, handle_cmd_complete},
-};
-
-static const size_t numof_commands = sizeof(commands) / sizeof(commands[0]);
-
 static int server_listen_thread(struct server *server)
 {
-	struct cmd_dispatcher *dispatcher = NULL;
 	int ret = 0;
 
-	ret = cmd_dispatcher_create(&dispatcher, commands, numof_commands, server);
-	if (ret < 0)
-		return ret;
-
-	while (!global_stop_flag) {
+	while (!server->stopping) {
 		log("Waiting for new connections\n");
 
-		ret = tcp_server_accept(server->tcp_server, cmd_dispatcher_handle_conn, dispatcher);
-		if (ret < 0) {
-			if ((errno == EINTR || errno == EINVAL) && global_stop_flag)
-				ret = 0;
-			goto dispatcher_destroy;
-		}
+		ret = event_loop_run(server->event_loop);
+		if (ret < 0)
+			return ret;
 	}
 
-dispatcher_destroy:
-	cmd_dispatcher_destroy(dispatcher);
-
-	return server_set_stopping(server);
+	return 0;
 }
 
 int server_main(struct server *server)
diff --git a/src/signal.c b/src/signal.c
index 7d00bc9..9f788b2 100644
--- a/src/signal.c
+++ b/src/signal.c
@@ -7,14 +7,25 @@
 
 #include "signal.h"
 #include "compiler.h"
+#include "event_loop.h"
+#include "file.h"
 #include "log.h"
 
 #include <signal.h>
 #include <stddef.h>
 #include <string.h>
+#include <sys/signalfd.h>
+#include <unistd.h>
 
 static int stop_signals[] = {SIGINT, SIGTERM, SIGQUIT};
 
+static void stops_set(sigset_t *set)
+{
+	sigemptyset(set);
+	for (size_t i = 0; i < sizeof(stop_signals) / sizeof(stop_signals[0]); ++i)
+		sigaddset(set, stop_signals[i]);
+}
+
 volatile sig_atomic_t global_stop_flag = 0;
 
 static void set_global_stop_flag(UNUSED int signum)
@@ -78,11 +89,7 @@ int signal_block_all(sigset_t *old)
 int signal_block_stops(void)
 {
 	sigset_t set;
-	sigemptyset(&set);
-
-	for (size_t i = 0; i < sizeof(stop_signals) / sizeof(stop_signals[0]); ++i)
-		sigaddset(&set, stop_signals[i]);
-
+	stops_set(&set);
 	return signal_set(&set, NULL);
 }
 
@@ -90,3 +97,43 @@ int signal_restore(const sigset_t *new)
 {
 	return signal_set(new, NULL);
 }
+
+int signalfd_create(const sigset_t *set)
+{
+	sigset_t old;
+	int ret = 0;
+
+	ret = signal_set(set, &old);
+	if (ret < 0)
+		return ret;
+
+	ret = signalfd(-1, set, SFD_CLOEXEC);
+	if (ret < 0)
+		goto restore;
+
+	return ret;
+
+restore:
+	signal_set(&old, NULL);
+
+	return ret;
+}
+
+int signalfd_listen_for_stops(void)
+{
+	sigset_t set;
+	stops_set(&set);
+	return signalfd_create(&set);
+}
+
+void signalfd_destroy(int fd)
+{
+	file_close(fd);
+}
+
+int signalfd_add_to_event_loop(int fd, struct event_loop *loop, event_loop_handler handler,
+                               void *arg)
+{
+	struct event_fd entry = {.fd = fd, .events = POLLIN, .handler = handler, .arg = arg};
+	return event_loop_add(loop, &entry);
+}
diff --git a/src/signal.h b/src/signal.h
index 4f1c280..e3f5897 100644
--- a/src/signal.h
+++ b/src/signal.h
@@ -8,6 +8,8 @@
 #ifndef __SIGNAL_H__
 #define __SIGNAL_H__
 
+#include "event_loop.h"
+
 #include <signal.h>
 
 extern volatile sig_atomic_t global_stop_flag;
@@ -18,4 +20,11 @@ int signal_block_stops(void);
 
 int signal_restore(const sigset_t *new);
 
+int signalfd_create(const sigset_t *);
+void signalfd_destroy(int fd);
+
+int signalfd_add_to_event_loop(int fd, struct event_loop *, event_loop_handler handler, void *arg);
+
+int signalfd_listen_for_stops(void);
+
 #endif
diff --git a/src/string.c b/src/string.c
new file mode 100644
index 0000000..878efeb
--- /dev/null
+++ b/src/string.c
@@ -0,0 +1,25 @@
+/*
+ * Copyright (c) 2023 Egor Tensin <Egor.Tensin@gmail.com>
+ * This file is part of the "cimple" project.
+ * For details, see https://github.com/egor-tensin/cimple.
+ * Distributed under the MIT License.
+ */
+
+#include "string.h"
+
+#include <string.h>
+
+char *stpecpy(char *dst, char *end, const char *src)
+{
+	if (!dst)
+		return NULL;
+	if (dst == end)
+		return end;
+
+	char *p = memccpy(dst, src, '\0', end - dst);
+	if (p)
+		return p - 1;
+
+	end[-1] = '\0';
+	return end;
+}
diff --git a/src/string.h b/src/string.h
new file mode 100644
index 0000000..fb77e64
--- /dev/null
+++ b/src/string.h
@@ -0,0 +1,13 @@
+/*
+ * Copyright (c) 2023 Egor Tensin <Egor.Tensin@gmail.com>
+ * This file is part of the "cimple" project.
+ * For details, see https://github.com/egor-tensin/cimple.
+ * Distributed under the MIT License.
+ */
+
+#ifndef __STRING_H__
+#define __STRING_H__
+
+char *stpecpy(char *dst, char *end, const char *src);
+
+#endif
diff --git a/src/tcp_server.c b/src/tcp_server.c
index 2e65bc5..a6263f7 100644
--- a/src/tcp_server.c
+++ b/src/tcp_server.c
@@ -6,6 +6,8 @@
  */
 
 #include "tcp_server.h"
+#include "compiler.h"
+#include "event_loop.h"
 #include "log.h"
 #include "net.h"
 #include "signal.h"
@@ -16,18 +18,24 @@
 
 struct tcp_server {
 	int fd;
+	tcp_server_conn_handler conn_handler;
+	void *arg;
 };
 
-int tcp_server_create(struct tcp_server **_server, const char *port)
+int tcp_server_create(struct tcp_server **_server, const char *port,
+                      tcp_server_conn_handler conn_handler, void *arg)
 {
 	int ret = 0;
 
-	struct tcp_server *server = malloc(sizeof(struct tcp_server));
+	struct tcp_server *server = calloc(1, sizeof(struct tcp_server));
 	if (!server) {
 		log_errno("malloc");
 		return -1;
 	}
 
+	server->conn_handler = conn_handler;
+	server->arg = arg;
+
 	ret = net_bind(port);
 	if (ret < 0)
 		goto free;
@@ -50,7 +58,7 @@ void tcp_server_destroy(struct tcp_server *server)
 
 struct child_context {
 	int fd;
-	tcp_server_conn_handler handler;
+	tcp_server_conn_handler conn_handler;
 	void *arg;
 };
 
@@ -65,7 +73,7 @@ static void *connection_thread(void *_ctx)
 	if (ret < 0)
 		goto free_ctx;
 
-	ctx->handler(ctx->fd, ctx->arg);
+	ctx->conn_handler(ctx->fd, ctx->arg);
 
 free_ctx:
 	free(ctx);
@@ -73,7 +81,7 @@ free_ctx:
 	return NULL;
 }
 
-int tcp_server_accept(const struct tcp_server *server, tcp_server_conn_handler handler, void *arg)
+static int create_connection_thread(int fd, tcp_server_conn_handler conn_handler, void *arg)
 {
 	sigset_t old_mask;
 	pthread_t child;
@@ -81,24 +89,20 @@ int tcp_server_accept(const struct tcp_server *server, tcp_server_conn_handler h
 
 	struct child_context *ctx = calloc(1, sizeof(*ctx));
 	if (!ctx) {
-		log_errno("malloc");
+		log_errno("calloc");
 		return -1;
 	}
 
-	ctx->handler = handler;
+	ctx->fd = fd;
+	ctx->conn_handler = conn_handler;
 	ctx->arg = arg;
 
-	ret = net_accept(server->fd);
-	if (ret < 0)
-		goto free_ctx;
-	ctx->fd = ret;
-
 	/* Block all signals (we'll unblock them later); the child thread will
 	 * have all signals blocked initially. This allows the main thread to
 	 * handle SIGINT/SIGTERM/etc. */
 	ret = signal_block_all(&old_mask);
 	if (ret < 0)
-		goto close_conn;
+		goto free_ctx;
 
 	ret = pthread_create(&child, NULL, connection_thread, ctx);
 	if (ret) {
@@ -106,19 +110,53 @@ int tcp_server_accept(const struct tcp_server *server, tcp_server_conn_handler h
 		goto restore_mask;
 	}
 
+restore_mask:
 	/* Restore the previously-enabled signals for handling in the main thread. */
 	signal_restore(&old_mask);
 
 	return ret;
 
-restore_mask:
-	signal_restore(&old_mask);
-
-close_conn:
-	net_close(ctx->fd);
-
 free_ctx:
 	free(ctx);
 
 	return ret;
 }
+
+int tcp_server_accept(const struct tcp_server *server)
+{
+	int fd = -1, ret = 0;
+
+	ret = net_accept(server->fd);
+	if (ret < 0)
+		return ret;
+	fd = ret;
+
+	ret = create_connection_thread(fd, server->conn_handler, server->arg);
+	if (ret < 0)
+		goto close_conn;
+
+	return ret;
+
+close_conn:
+	net_close(fd);
+
+	return ret;
+}
+
+static int tcp_server_event_loop_handler(UNUSED struct event_loop *loop, UNUSED int fd,
+                                         UNUSED short revents, void *_server)
+{
+	struct tcp_server *server = (struct tcp_server *)_server;
+	return tcp_server_accept(server);
+}
+
+int tcp_server_add_to_event_loop(struct tcp_server *server, struct event_loop *loop)
+{
+	struct event_fd entry = {
+	    .fd = server->fd,
+	    .events = POLLIN,
+	    .handler = tcp_server_event_loop_handler,
+	    .arg = server,
+	};
+	return event_loop_add(loop, &entry);
+}
diff --git a/src/tcp_server.h b/src/tcp_server.h
index 0f377d8..f96e3f1 100644
--- a/src/tcp_server.h
+++ b/src/tcp_server.h
@@ -8,12 +8,17 @@
 #ifndef __TCP_SERVER_H__
 #define __TCP_SERVER_H__
 
+#include "event_loop.h"
+
 struct tcp_server;
 
-int tcp_server_create(struct tcp_server **, const char *port);
+typedef int (*tcp_server_conn_handler)(int conn_fd, void *arg);
+
+int tcp_server_create(struct tcp_server **, const char *port, tcp_server_conn_handler, void *arg);
 void tcp_server_destroy(struct tcp_server *);
 
-typedef int (*tcp_server_conn_handler)(int conn_fd, void *arg);
-int tcp_server_accept(const struct tcp_server *, tcp_server_conn_handler, void *arg);
+int tcp_server_accept(const struct tcp_server *);
+
+int tcp_server_add_to_event_loop(struct tcp_server *, struct event_loop *);
 
 #endif
diff --git a/src/worker.c b/src/worker.c
index d7ad310..8ecaed8 100644
--- a/src/worker.c
+++ b/src/worker.c
@@ -10,74 +10,78 @@
 #include "command.h"
 #include "compiler.h"
 #include "const.h"
+#include "event_loop.h"
 #include "git.h"
 #include "log.h"
 #include "msg.h"
+#include "net.h"
 #include "process.h"
 #include "run_queue.h"
 #include "signal.h"
 
 #include <stdlib.h>
+#include <string.h>
 
 struct worker {
-	int dummy;
-};
+	struct settings *settings;
 
-int worker_create(struct worker **_worker)
-{
-	int ret = 0;
+	int stopping;
 
-	ret = signal_handle_stops();
-	if (ret < 0)
-		return ret;
+	struct cmd_dispatcher *cmd_dispatcher;
 
-	struct worker *worker = malloc(sizeof(struct worker));
-	if (!worker) {
+	struct event_loop *event_loop;
+	int signalfd;
+};
+
+static struct settings *worker_settings_copy(const struct settings *src)
+{
+	struct settings *result = malloc(sizeof(*src));
+	if (!result) {
 		log_errno("malloc");
-		return -1;
+		return NULL;
 	}
 
-	ret = libgit_init();
-	if (ret < 0)
-		goto free;
-
-	*_worker = worker;
-	return ret;
+	result->host = strdup(src->host);
+	if (!result->host) {
+		log_errno("strdup");
+		goto free_result;
+	}
 
-free:
-	free(worker);
+	result->port = strdup(src->port);
+	if (!result->port) {
+		log_errno("strdup");
+		goto free_host;
+	}
 
-	return ret;
-}
+	return result;
 
-void worker_destroy(struct worker *worker)
-{
-	log("Shutting down\n");
+free_host:
+	free(result->host);
 
-	libgit_shutdown();
-	free(worker);
-}
+free_result:
+	free(result);
 
-static int worker_send_to_server(const struct settings *settings, const struct msg *request,
-                                 struct msg **response)
-{
-	return msg_connect_and_talk(settings->host, settings->port, request, response);
+	return NULL;
 }
 
-static int worker_send_to_server_argv(const struct settings *settings, const char **argv,
-                                      struct msg **response)
+static void worker_settings_destroy(struct settings *settings)
 {
-	return msg_connect_and_talk_argv(settings->host, settings->port, argv, response);
+	free(settings->port);
+	free(settings->host);
+	free(settings);
 }
 
-static int worker_send_new_worker(const struct settings *settings, struct msg **task)
+static int worker_set_stopping(UNUSED struct event_loop *loop, UNUSED int fd, UNUSED short revents,
+                               void *_worker)
 {
-	static const char *argv[] = {CMD_NEW_WORKER, NULL};
-	return worker_send_to_server_argv(settings, argv, task);
+	struct worker *worker = (struct worker *)_worker;
+	worker->stopping = 1;
+	return 0;
 }
 
-static int msg_run_handler(const struct msg *request, struct msg **response, UNUSED void *ctx)
+static int worker_handle_run(const struct msg *request, UNUSED struct msg **response, void *_ctx)
 {
+	struct cmd_conn_ctx *ctx = (struct cmd_conn_ctx *)_ctx;
 	struct run *run = NULL;
 	struct proc_output result;
 	int ret = 0;
@@ -96,11 +100,17 @@ static int msg_run_handler(const struct msg *request, struct msg **response, UNU
 
 	proc_output_dump(&result);
 
+	struct worker *worker = (struct worker *)ctx->arg;
 	static const char *argv[] = {CMD_COMPLETE, NULL};
-	ret = msg_from_argv(response, argv);
+
+	ret = msg_connect_and_talk_argv(worker->settings->host, worker->settings->port, argv, NULL);
 	if (ret < 0)
 		goto free_output;
 
+	/* Close the descriptor and remove it from the event loop.
+	 * poll(2) is too confusing, honestly. */
+	ret = EVENT_LOOP_REMOVE;
+
 free_output:
 	proc_output_free(&result);
 
@@ -110,49 +120,111 @@ free_output:
 }
 
 static struct cmd_desc commands[] = {
-    {CMD_RUN, msg_run_handler},
+    {CMD_RUN, worker_handle_run},
 };
 
 static const size_t numof_commands = sizeof(commands) / sizeof(commands[0]);
 
-int worker_main(UNUSED struct worker *worker, const struct settings *settings)
+int worker_create(struct worker **_worker, const struct settings *settings)
 {
-	struct msg *task = NULL;
-	struct cmd_dispatcher *dispatcher = NULL;
 	int ret = 0;
 
-	ret = cmd_dispatcher_create(&dispatcher, commands, numof_commands, NULL);
+	struct worker *worker = malloc(sizeof(struct worker));
+	if (!worker) {
+		log_errno("malloc");
+		return -1;
+	}
+
+	worker->settings = worker_settings_copy(settings);
+	if (!worker->settings) {
+		ret = -1;
+		goto free;
+	}
+
+	worker->stopping = 0;
+
+	ret = cmd_dispatcher_create(&worker->cmd_dispatcher, commands, numof_commands, worker);
 	if (ret < 0)
-		return ret;
+		goto free_settings;
 
-	log("Waiting for a new command\n");
+	ret = event_loop_create(&worker->event_loop);
+	if (ret < 0)
+		goto destroy_cmd_dispatcher;
 
-	ret = worker_send_new_worker(settings, &task);
-	if (ret < 0) {
-		if ((errno == EINTR || errno == EINVAL) && global_stop_flag)
-			ret = 0;
-		goto dispatcher_destroy;
-	}
+	ret = signalfd_listen_for_stops();
+	if (ret < 0)
+		goto destroy_event_loop;
+	worker->signalfd = ret;
+
+	ret = signalfd_add_to_event_loop(worker->signalfd, worker->event_loop, worker_set_stopping,
+	                                 worker);
+	if (ret < 0)
+		goto close_signalfd;
+
+	ret = libgit_init();
+	if (ret < 0)
+		goto close_signalfd;
+
+	*_worker = worker;
+	return ret;
+
+close_signalfd:
+	signalfd_destroy(worker->signalfd);
+
+destroy_event_loop:
+	event_loop_destroy(worker->event_loop);
+
+destroy_cmd_dispatcher:
+	cmd_dispatcher_destroy(worker->cmd_dispatcher);
+
+free_settings:
+	worker_settings_destroy(worker->settings);
+
+free:
+	free(worker);
+
+	return ret;
+}
 
-	while (!global_stop_flag) {
-		struct msg *result = NULL;
+void worker_destroy(struct worker *worker)
+{
+	log("Shutting down\n");
 
-		ret = cmd_dispatcher_handle(dispatcher, task, &result);
-		msg_free(task);
+	libgit_shutdown();
+	signalfd_destroy(worker->signalfd);
+	event_loop_destroy(worker->event_loop);
+	cmd_dispatcher_destroy(worker->cmd_dispatcher);
+	worker_settings_destroy(worker->settings);
+	free(worker);
+}
+
+int worker_main(struct worker *worker)
+{
+	int ret = 0;
+
+	while (!worker->stopping) {
+		ret = net_connect(worker->settings->host, worker->settings->port);
 		if (ret < 0)
-			goto dispatcher_destroy;
-
-		ret = worker_send_to_server(settings, result, &task);
-		msg_free(result);
-		if (ret < 0) {
-			if ((errno == EINTR || errno == EINVAL) && global_stop_flag)
-				ret = 0;
-			goto dispatcher_destroy;
-		}
-	}
+			return ret;
+
+		const int fd = ret;
+		static const char *argv[] = {CMD_NEW_WORKER, NULL};
+
+		ret = msg_send_argv(fd, argv);
+		if (ret < 0)
+			return ret;
 
-dispatcher_destroy:
-	cmd_dispatcher_destroy(dispatcher);
+		ret = cmd_dispatcher_add_to_event_loop(worker->cmd_dispatcher, worker->event_loop,
+		                                       fd);
+		if (ret < 0)
+			return ret;
+
+		log("Waiting for a new command\n");
+
+		ret = event_loop_run(worker->event_loop);
+		if (ret < 0)
+			return ret;
+	}
 
 	return ret;
 }
diff --git a/src/worker.h b/src/worker.h
index bf603df..8eb325c 100644
--- a/src/worker.h
+++ b/src/worker.h
@@ -15,9 +15,9 @@ struct settings {
 
 struct worker;
 
-int worker_create(struct worker **);
+int worker_create(struct worker **, const struct settings *);
 void worker_destroy(struct worker *);
 
-int worker_main(struct worker *, const struct settings *);
+int worker_main(struct worker *);
 
 #endif
diff --git a/src/worker_main.c b/src/worker_main.c
index b9e9b9b..dcfee47 100644
--- a/src/worker_main.c
+++ b/src/worker_main.c
@@ -70,11 +70,11 @@ int main(int argc, char *argv[])
 	if (ret < 0)
 		return ret;
 
-	ret = worker_create(&worker);
+	ret = worker_create(&worker, &settings);
 	if (ret < 0)
 		return ret;
 
-	ret = worker_main(worker, &settings);
+	ret = worker_main(worker);
 	if (ret < 0)
 		goto destroy_worker;