From 2d6b6d4da1d9732b2b02989a4374d356ff840d1b Mon Sep 17 00:00:00 2001 From: Dirk Engling Date: Mon, 26 Oct 2026 20:03:22 +0100 Subject: Replace libowfat's event waiter with our own abstraction It's with a heavy heart that I remove libowfat as my event library from opentracker. libowfat is what taught opentracker to walk and then fly. The tracker was literally standing on the shoulder of giants when I started writing it in 2006. For its time, the event abstraction was unparalleled and it has held up for many years, and it served the bittorrent community greatly. https://erdgeist.org/gitweb/opentracker/commit/?id=ad472597c569568bad0526d116b601dfcb36638c Unfortunately, it is very likely that libowfat will not be maintained anymore for the foreseeable future, it is already not easy to get its source from upstream and opentracker started receiving bug reports that I can not reasonably answer or patch. Before the project gets more dependent on the limitations of libowfat's current state, I need to teach opentracker to stand on its own feet and fly by itself. I've learned a great deal in the process and have even more respect now for what libowfat pioneered. The path for opentracker is cleared now. Ground work has already been laid in the migration process to support multi threaded http processing while keeping the aggressive reduction of syscalls alive that made the libowfat+opentracker combo so scalable. Here be some dragons, though: Even though there is a unit test suite now, the event loop's rewrite has not been battle tested yet. So you might want to hold up for a bit before jumping on the solo band wagon. What awaits at the end will be a long overdue v1.0, which will be named in honour of Felix. --- opentracker.c | 95 +++++++++++++++++++++++++++++++++++------------------------ 1 file changed, 57 insertions(+), 38 deletions(-) (limited to 'opentracker.c') diff --git a/opentracker.c b/opentracker.c index 7de6c6a..21eeb42 100644 --- a/opentracker.c +++ b/opentracker.c @@ -21,9 +21,6 @@ #include #endif -/* Libowfat */ -#include "io.h" - /* Opentracker */ #include "ot_accesslist.h" #include "ot_config.h" @@ -34,6 +31,7 @@ #include "ot_mutex.h" #include "ot_random.h" #include "ot_scan.h" +#include "ot_event.h" #include "ot_fdstate.h" #include "ot_socket.h" #include "ot_stats.h" @@ -130,14 +128,15 @@ static void handle_dead(const int64_t sock) { } /* The generation has to move before the kernel can reissue the number */ fdstate_close(sock); - io_close(sock); + event_del((int)sock); + close((int)sock); } static void handle_read(const int64_t sock, struct ot_workstruct *ws) { struct http_data *cookie = fdstate_cookie(sock); - ssize_t byte_count = io_tryread(sock, ws->inbuf, G_INBUF_SIZE); + ssize_t byte_count = read((int)sock, ws->inbuf, G_INBUF_SIZE); - if (byte_count == 0 || byte_count == -3) { + if (byte_count == 0 || (byte_count == -1 && errno != EAGAIN && errno != EWOULDBLOCK && errno != EINTR)) { handle_dead(sock); return; } @@ -197,7 +196,7 @@ static void handle_write(const int64_t sock) { /* In a chunked transfer, wait for the next piece once everything queued has gone out */ if (chunked) - io_dontwantwrite(sock); + http_interest_drop(sock, cookie, EVENT_OUT); else handle_dead(sock); } @@ -210,25 +209,30 @@ static void handle_accept(const int64_t serversocket) { while ((sock = socket_accept(serversocket, ip, &port)) != -1) { - /* Put fd into a non-blocking mode */ - io_nonblock(sock); + socket_set_blocking((int)sock, 0); - if (!io_fd(sock) || !(cookie = (struct http_data *)malloc(sizeof(struct http_data)))) { - io_close(sock); + if (!(cookie = (struct http_data *)malloc(sizeof(struct http_data)))) { + close((int)sock); continue; } memset(cookie, 0, sizeof(struct http_data)); memcpy(cookie->ip, ip, sizeof(ot_ip6)); + cookie->interest = EVENT_IN; - fdstate_open(sock, cookie, g_now_seconds + OT_CLIENT_TIMEOUT); - io_wantread(sock); + if (event_add((int)sock, EVENT_IN, fdstate_open(sock, cookie, g_now_seconds + OT_CLIENT_TIMEOUT)) == -1) { + fdstate_close(sock); + free(cookie); + close((int)sock); + continue; + } stats_issue_event(EVENT_ACCEPT, FLAG_TCP, (uintptr_t)ip); - } - io_eagain(serversocket); } +/* How many ready descriptors one wait hands back at a time */ +#define OT_EVENT_BATCH 512 + static void *server_mainloop(void *args) { struct ot_workstruct ws; time_t next_timeout_check = g_now_seconds + OT_CLIENT_TIMEOUT_CHECKINTERVAL; @@ -251,28 +255,42 @@ static void *server_mainloop(void *args) { random_bytes(ws.rand48_state, sizeof(ws.rand48_state)); for (;;) { - int64_t sock; + ot_event ready[OT_EVENT_BATCH]; + int64_t sock; + int count, i; - io_wait(); + count = event_wait(ready, OT_EVENT_BATCH, OT_CLIENT_TIMEOUT_CHECKINTERVAL * 1000); + + for (i = 0; i < count; ++i) { + const void *cookie; + + sock = ready[i].fd; + /* A descriptor closed earlier in this same batch may already have been + handed out again by accept, so an event is only ours if the + generation it was registered under is still the one in place */ + if (fdstate_generation((int)sock) != ready[i].generation) + continue; + cookie = fdstate_cookie((int)sock); - while ((sock = io_canread()) != -1) { - const void *cookie = fdstate_cookie(sock); if ((intptr_t)cookie == FLAG_TCP) handle_accept(sock); else if ((intptr_t)cookie == FLAG_UDP) handle_udp6(sock, &ws); else if ((intptr_t)cookie == FLAG_SELFPIPE) - io_tryread(sock, ws.inbuf, G_INBUF_SIZE); - else - handle_read(sock, &ws); + while (read((int)sock, ws.inbuf, G_INBUF_SIZE) > 0) + ; + else { + if (ready[i].what & EVENT_IN) + handle_read(sock, &ws); + /* The read may have closed it, so look again before writing */ + if ((ready[i].what & EVENT_OUT) && fdstate_generation((int)sock) == ready[i].generation) + handle_write(sock); + } } while ((sock = mutex_workqueue_popresult(&iovec_entries, &iovector, &is_partial)) != -1) http_sendiovecdata(sock, &ws, iovec_entries, iovector, is_partial); - while ((sock = io_canwrite()) != -1) - handle_write(sock); - if (g_now_seconds > next_timeout_check) { int cursor = 0; @@ -319,11 +337,8 @@ static void listen_bind_one(ot_listen *entry) { panic("socket_udp_bind"); } - if (!io_fd(entry->sock)) - panic("io_fd"); - - fdstate_open(entry->sock, (void *)entry->proto, 0); - io_wantread(entry->sock); + if (event_add((int)entry->sock, EVENT_IN, fdstate_open(entry->sock, (void *)entry->proto, 0)) == -1) + panic("event_add"); #ifdef _DEBUG fputs(" success.\n", stderr); @@ -352,7 +367,8 @@ static void listen_start_udp_workers(void) { if (entry->proto != FLAG_UDP || entry->workers <= 0) continue; - io_block(entry->sock); + socket_set_blocking((int)entry->sock, 1); + event_del((int)entry->sock); udp_workers_install(entry->sock, (unsigned int)entry->workers); } } @@ -453,6 +469,8 @@ int main(int argc, char **argv) { random_init(); if (fdstate_init() == -1) panic("fdstate_init"); + if (event_init() == -1) + panic("event_init"); config_parse(argc, argv); listen_bind_all(); @@ -471,15 +489,15 @@ int main(int argc, char **argv) { pthread_create(&thread_id, NULL, time_caching_worker, NULL); /* Create our self pipe which allows us to interrupt mainloops - io_wait in case some data is available to send out */ + wait in case some data is available to send out */ if (pipe(g_self_pipe) == -1) panic("selfpipe failed: "); - if (!io_fd(g_self_pipe[0])) - panic("selfpipe io_fd failed: "); - if (!io_fd(g_self_pipe[1])) - panic("selfpipe io_fd failed: "); - fdstate_open(g_self_pipe[0], (void *)FLAG_SELFPIPE, 0); - io_wantread(g_self_pipe[0]); + /* Both ends have to be non blocking: the loop drains the read end until it + is empty, and a worker nudging a full pipe has already made its point */ + socket_set_blocking(g_self_pipe[0], 0); + socket_set_blocking(g_self_pipe[1], 0); + if (event_add(g_self_pipe[0], EVENT_IN, fdstate_open(g_self_pipe[0], (void *)FLAG_SELFPIPE, 0)) == -1) + panic("selfpipe event_add"); /* Init all sub systems. This call may fail with an exit() */ trackerlogic_init(); @@ -499,6 +517,7 @@ int main(int argc, char **argv) { server_mainloop(0); + event_deinit(); fdstate_deinit(); return 0; } -- cgit v1.3