From 5e86efde052760a53ca92ddd411db095fcf4a970 Mon Sep 17 00:00:00 2001 From: Dirk Engling Date: Mon, 26 Oct 2026 16:37:55 +0100 Subject: Use new fdstate subsystem to store per fd cookie and timeout --- Makefile | 2 +- opentracker.c | 26 ++++++++++++++++---------- ot_fdstate.c | 28 ++++++++++++++++++++++++++-- ot_fdstate.h | 5 +++++ ot_http.c | 24 ++++++++++++------------ ot_timeout.h | 39 --------------------------------------- proxy.c | 15 +++++++++------ 7 files changed, 69 insertions(+), 70 deletions(-) delete mode 100644 ot_timeout.h diff --git a/Makefile b/Makefile index 3bf254a..87c20e2 100644 --- a/Makefile +++ b/Makefile @@ -70,7 +70,7 @@ LDFLAGS+=-L$(LIBOWFAT_LIBRARY) -lowfat -pthread BINARY =opentracker HEADERS=trackerlogic.h scan_urlencoded_query.h ot_batch.h ot_buf.h ot_fdstate.h ot_hex.h ot_ip.h ot_config.h ot_scan.h ot_socket.h ot_mutex.h ot_random.h ot_stats.h ot_vector.h ot_clean.h ot_udp.h ot_iovec.h ot_fullscrape.h ot_accesslist.h ot_http.h ot_livesync.h ot_rijndael.h SOURCES=opentracker.c trackerlogic.c scan_urlencoded_query.c ot_batch.c ot_buf.c ot_fdstate.c ot_hex.c ot_ip.c ot_config.c ot_scan.c ot_socket.c ot_mutex.c ot_random.c ot_stats.c ot_vector.c ot_clean.c ot_udp.c ot_iovec.c ot_fullscrape.c ot_accesslist.c ot_http.c ot_livesync.c ot_rijndael.c -SOURCES_proxy=proxy.c ot_batch.c ot_vector.c ot_mutex.c ot_random.c ot_scan.c ot_ip.c ot_socket.c +SOURCES_proxy=proxy.c ot_batch.c ot_fdstate.c ot_vector.c ot_mutex.c ot_random.c ot_scan.c ot_ip.c ot_socket.c OBJECTS = $(SOURCES:%.c=%.o) OBJECTS_debug = $(SOURCES:%.c=%.debug.o) diff --git a/opentracker.c b/opentracker.c index 60bdceb..0f0b314 100644 --- a/opentracker.c +++ b/opentracker.c @@ -34,8 +34,8 @@ #include "ot_mutex.h" #include "ot_random.h" #include "ot_scan.h" +#include "ot_fdstate.h" #include "ot_socket.h" -#include "ot_timeout.h" #include "ot_stats.h" #include "ot_udp.h" #include "trackerlogic.h" @@ -118,7 +118,7 @@ static ssize_t header_complete(char *request, ssize_t byte_count) { } static void handle_dead(const int64_t sock) { - struct http_data *cookie = io_getcookie(sock); + struct http_data *cookie = fdstate_cookie(sock); if (cookie) { if (cookie->batch) batch_reset(cookie->batch); @@ -128,11 +128,13 @@ static void handle_dead(const int64_t sock) { mutex_workqueue_canceltask(sock); free(cookie); } + /* The generation has to move before the kernel can reissue the number */ + fdstate_close(sock); io_close(sock); } static void handle_read(const int64_t sock, struct ot_workstruct *ws) { - struct http_data *cookie = io_getcookie(sock); + struct http_data *cookie = fdstate_cookie(sock); ssize_t byte_count = io_tryread(sock, ws->inbuf, G_INBUF_SIZE); if (byte_count == 0 || byte_count == -3) { @@ -176,7 +178,7 @@ static void handle_read(const int64_t sock, struct ot_workstruct *ws) { } static void handle_write(const int64_t sock) { - struct http_data *cookie = io_getcookie(sock); + struct http_data *cookie = fdstate_cookie(sock); int chunked = 0; if (cookie) { @@ -218,12 +220,11 @@ static void handle_accept(const int64_t serversocket) { memset(cookie, 0, sizeof(struct http_data)); memcpy(cookie->ip, ip, sizeof(ot_ip6)); - io_setcookie(sock, cookie); + fdstate_open(sock, cookie, g_now_seconds + OT_CLIENT_TIMEOUT); io_wantread(sock); stats_issue_event(EVENT_ACCEPT, FLAG_TCP, (uintptr_t)ip); - timeout_in(sock, OT_CLIENT_TIMEOUT); } io_eagain(serversocket); } @@ -255,7 +256,7 @@ static void *server_mainloop(void *args) { io_wait(); while ((sock = io_canread()) != -1) { - const void *cookie = io_getcookie(sock); + const void *cookie = fdstate_cookie(sock); if ((intptr_t)cookie == FLAG_TCP) handle_accept(sock); else if ((intptr_t)cookie == FLAG_UDP) @@ -273,7 +274,9 @@ static void *server_mainloop(void *args) { handle_write(sock); if (g_now_seconds > next_timeout_check) { - while ((sock = io_timeouted()) != -1) + int cursor = 0; + + while ((sock = fdstate_next_expired(g_now_seconds, &cursor)) != -1) handle_dead(sock); next_timeout_check = g_now_seconds + OT_CLIENT_TIMEOUT_CHECKINTERVAL; } @@ -319,7 +322,7 @@ static void listen_bind_one(ot_listen *entry) { if (!io_fd(entry->sock)) panic("io_fd"); - io_setcookie(entry->sock, (void *)entry->proto); + fdstate_open(entry->sock, (void *)entry->proto, 0); io_wantread(entry->sock); #ifdef _DEBUG @@ -448,6 +451,8 @@ int main(int argc, char **argv) { pthread_t thread_id; /* time cacher */ random_init(); + if (fdstate_init() == -1) + panic("fdstate_init"); config_parse(argc, argv); listen_bind_all(); @@ -473,7 +478,7 @@ int main(int argc, char **argv) { panic("selfpipe io_fd failed: "); if (!io_fd(g_self_pipe[1])) panic("selfpipe io_fd failed: "); - io_setcookie(g_self_pipe[0], (void *)FLAG_SELFPIPE); + fdstate_open(g_self_pipe[0], (void *)FLAG_SELFPIPE, 0); io_wantread(g_self_pipe[0]); /* Init all sub systems. This call may fail with an exit() */ @@ -494,5 +499,6 @@ int main(int argc, char **argv) { server_mainloop(0); + fdstate_deinit(); return 0; } diff --git a/ot_fdstate.c b/ot_fdstate.c index 6ddbea3..7f73d3d 100644 --- a/ot_fdstate.c +++ b/ot_fdstate.c @@ -34,8 +34,11 @@ typedef struct { range descriptor is rejected */ #define FD_SLOTS_MAX (1u << 20) -static ot_fdslot *g_slots; -static size_t g_slot_count; +static ot_fdslot *g_slots; +static size_t g_slot_count; +/* Descriptors are handed out lowest first, so a sweep only has to walk as + far as the highest one ever seen rather than the whole table. */ +static _Atomic size_t g_slot_highest; int fdstate_init(void) { struct rlimit limit; @@ -51,6 +54,7 @@ int fdstate_init(void) { } g_slot_count = count; + atomic_store(&g_slot_highest, 0); return 0; } @@ -59,6 +63,7 @@ void fdstate_deinit(void) { munmap(g_slots, g_slot_count * sizeof(ot_fdslot)); g_slots = NULL; g_slot_count = 0; + atomic_store(&g_slot_highest, 0); } size_t fdstate_slots(void) { return g_slot_count; } @@ -76,6 +81,15 @@ uint32_t fdstate_open(int fd, void *cookie, ot_time expiry) { if (!slot) return 0; + { + size_t highest = atomic_load_explicit(&g_slot_highest, memory_order_relaxed); + + while ((size_t)fd > highest && + !atomic_compare_exchange_weak_explicit(&g_slot_highest, &highest, (size_t)fd, memory_order_relaxed, + memory_order_relaxed)) + ; + } + word = atomic_load_explicit(&slot->state, memory_order_relaxed); atomic_store_explicit(&slot->cookie, cookie, memory_order_relaxed); atomic_store_explicit(&slot->expiry, expiry, memory_order_relaxed); @@ -150,6 +164,16 @@ void fdstate_close(int fd) { atomic_store_explicit(&slot->state, FD_WORD(FD_GEN(word) + 1, FD_FREE), memory_order_release); } +int fdstate_next_expired(ot_time now, int *cursor) { + size_t highest = atomic_load_explicit(&g_slot_highest, memory_order_relaxed); + + for (; (size_t)*cursor <= highest; ++*cursor) + if (fdstate_claim_expired(*cursor, now)) + return (*cursor)++; + + return -1; +} + void *fdstate_cookie(int fd) { ot_fdslot *slot = fdstate_slot(fd); diff --git a/ot_fdstate.h b/ot_fdstate.h index 2693a3b..88f685d 100644 --- a/ot_fdstate.h +++ b/ot_fdstate.h @@ -45,6 +45,11 @@ int fdstate_claim_expired(int fd, ot_time now); quoting the old generation are still in flight. */ void fdstate_close(int fd); +/* Walks the table for descriptors that have expired, handing back one at a + time until it returns -1. The cursor belongs to the caller, so several + sweepers can run without sharing anything. */ +int fdstate_next_expired(ot_time now, int *cursor); + /* Both only valid while the caller holds the descriptor. */ void *fdstate_cookie(int fd); void fdstate_set_expiry(int fd, ot_time expiry); diff --git a/ot_http.c b/ot_http.c index 331a457..6bf343c 100644 --- a/ot_http.c +++ b/ot_http.c @@ -25,8 +25,8 @@ #include "ot_ip.h" #include "ot_mutex.h" #include "ot_scan.h" +#include "ot_fdstate.h" #include "ot_stats.h" -#include "ot_timeout.h" #include "scan_urlencoded_query.h" #include "trackerlogic.h" @@ -39,7 +39,7 @@ ssize_t g_stats_path_len; enum { SUCCESS_HTTP_HEADER_LENGTH = 80, SUCCESS_HTTP_SIZE_OFF = 17 }; static void http_senddata(const int64_t sock, struct ot_workstruct *ws) { - struct http_data *cookie = io_getcookie(sock); + struct http_data *cookie = fdstate_cookie(sock); ssize_t written_size; if (!cookie) { @@ -97,7 +97,7 @@ static void http_senddata(const int64_t sock, struct ot_workstruct *ws) { /* writeable short data sockets just have a tcp timeout */ if (!ws->keep_alive) { - timeout_never(sock); + fdstate_set_expiry(sock, 0); io_dontwantread(sock); } io_wantwrite(sock); @@ -134,7 +134,7 @@ ssize_t http_issue_error(const int64_t sock, struct ot_workstruct *ws, int code) } ssize_t http_sendiovecdata(const int64_t sock, struct ot_workstruct *ws, size_t iovec_entries, struct iovec *iovector, int is_partial) { - struct http_data *cookie = io_getcookie(sock); + struct http_data *cookie = fdstate_cookie(sock); char *header; const char *encoding = ""; size_t header_size, size = iovec_length(&iovec_entries, (const struct iovec **)&iovector); @@ -207,7 +207,7 @@ ssize_t http_sendiovecdata(const int64_t sock, struct ot_workstruct *ws, size_t cookie->flag &= ~STRUCT_HTTP_FLAG_CHUNKED_IN_TRANSFER; } - timeout_in(sock, OT_CLIENT_TIMEOUT_SEND); + fdstate_set_expiry(sock, g_now_seconds + OT_CLIENT_TIMEOUT_SEND); io_wantwrite(sock); return 0; } @@ -243,7 +243,7 @@ static ssize_t http_handle_stats(const int64_t sock, struct ot_workstruct *ws, c {"url", TASK_FULLSCRAPE_TPB_URLENCODED}, {"txt", TASK_FULLSCRAPE_TPB_ASCII}, {"txtp", TASK_FULLSCRAPE_TPB_ASCII_PLUS}, {NULL, -3}}; - struct http_data *cookie = io_getcookie(sock); + struct http_data *cookie = fdstate_cookie(sock); int mode = TASK_STATS_PEERS, scanon = 1, format = 0; /* A socket without one is a dead socket, which is what http_sendiovecdata @@ -301,7 +301,7 @@ static ssize_t http_handle_stats(const int64_t sock, struct ot_workstruct *ws, c cookie->flag |= STRUCT_HTTP_FLAG_WAITINGFORTASK | STRUCT_HTTP_FLAG_CHUNKED; /* Clients waiting for us should not easily timeout */ - timeout_never(sock); + fdstate_set_expiry(sock, 0); fullscrape_deliver(sock, format); io_dontwantread(sock); return ws->reply_size = -2; @@ -314,7 +314,7 @@ static ssize_t http_handle_stats(const int64_t sock, struct ot_workstruct *ws, c cookie->flag |= STRUCT_HTTP_FLAG_WAITINGFORTASK; /* Complex stats also include expensive memory debugging tools */ - timeout_never(sock); + fdstate_set_expiry(sock, 0); stats_deliver(sock, mode); return ws->reply_size = -2; } @@ -334,7 +334,7 @@ typedef struct { #ifdef WANT_FULLSCRAPE static ssize_t http_handle_fullscrape(const int64_t sock, struct ot_workstruct *ws) { - struct http_data *cookie = io_getcookie(sock); + struct http_data *cookie = fdstate_cookie(sock); int format = 0; #ifdef WANT_MODEST_FULLSCRAPES @@ -394,7 +394,7 @@ static ssize_t http_handle_fullscrape(const int64_t sock, struct ot_workstruct * /* Pass this task to the worker thread */ cookie->flag |= STRUCT_HTTP_FLAG_WAITINGFORTASK | STRUCT_HTTP_FLAG_CHUNKED; /* Clients waiting for us should not easily timeout */ - timeout_never(sock); + fdstate_set_expiry(sock, 0); fullscrape_deliver(sock, format); io_dontwantread(sock); return ws->reply_size = -2; @@ -486,7 +486,7 @@ static ssize_t http_handle_announce(const int64_t sock, struct ot_workstruct unsigned short port = 0; char *write_ptr; ssize_t len; - struct http_data *cookie = io_getcookie(sock); + struct http_data *cookie = fdstate_cookie(sock); /* This is to hack around stupid clients that send "announce ?info_hash" */ if (read_ptr[-1] != '?') { @@ -663,7 +663,7 @@ ssize_t http_handle_request(const int64_t sock, struct ot_workstruct *ws) { char *read_ptr = ws->request, *write_ptr; #ifdef WANT_FULLLOG_NETWORKS - struct http_data *cookie = io_getcookie(sock); + struct http_data *cookie = fdstate_cookie(sock); if (loglist_check_address(cookie->ip)) { ot_log *log = malloc(sizeof(ot_log)); if (log) { diff --git a/ot_timeout.h b/ot_timeout.h deleted file mode 100644 index 1027c2e..0000000 --- a/ot_timeout.h +++ /dev/null @@ -1,39 +0,0 @@ -/* This software was written by Dirk Engling - It is considered beerware. Prost. Skol. Cheers or whatever. - - $id$ */ - -#ifndef OT_TIMEOUT_H__ -#define OT_TIMEOUT_H__ - -#include -#include - -/* Libowfat */ -#include "io.h" - -#include "trackerlogic.h" - -/* io_timeout wants an absolute TAI64 deadline and io_timeouted skips any - socket whose deadline is zero, so that is how a socket is told to wait - indefinitely. - - The deadline is taken from the cached clock rather than from - gettimeofday, which is too expensive to call at this rate on FreeBSD. */ - -static inline void timeout_never(int64_t sock) { - tai6464 t; - - memset(&t, 0, sizeof(t)); - io_timeout(sock, t); -} - -static inline void timeout_in(int64_t sock, unsigned int seconds) { - tai6464 t; - - memset(&t, 0, sizeof(t)); - tai_unix(&t.sec, g_now_seconds + seconds); - io_timeout(sock, t); -} - -#endif diff --git a/proxy.c b/proxy.c index 63b4c4f..9486e75 100644 --- a/proxy.c +++ b/proxy.c @@ -19,6 +19,7 @@ /* Opentracker */ #include "ot_batch.h" +#include "ot_fdstate.h" #include "ot_ip.h" #include "ot_mutex.h" #include "ot_random.h" @@ -267,7 +268,7 @@ static void handle_reconnects(void) { continue; /* No socket for you */ io_fd(newfd); io_wantwrite(newfd); /* So we will be informed when it is connected */ - io_setcookie(newfd, g_connections + i); + fdstate_open(newfd, g_connections + i, 0); /* Prepare connection info block */ reset_info_block(g_connections + i); @@ -310,7 +311,7 @@ static void handle_accept(int64_t serversocket) { g_connections[i].port = port; g_connections[i].fd = newfd; - io_setcookie(newfd, g_connections + i); + fdstate_open(newfd, g_connections + i, 0); /* We expect the connecting side to begin with its tracker_id */ io_wantread(newfd); @@ -325,7 +326,7 @@ static void handle_read(int64_t peersocket) { int i; int64_t datalen; uint32_t tracker_id; - proxy_peer *peer = io_getcookie(peersocket); + proxy_peer *peer = fdstate_cookie(peersocket); if (!peer) { /* Can't happen ;) */ @@ -391,7 +392,7 @@ static void handle_read(int64_t peersocket) { /* Can write new sync data to the stream */ static void handle_write(int64_t peersocket) { - proxy_peer *peer = io_getcookie(peersocket); + proxy_peer *peer = fdstate_cookie(peersocket); if (!peer) { /* Can't happen ;) */ @@ -464,7 +465,7 @@ static void server_mainloop() { /* Loop over readable sockets */ while ((sock = io_canread()) != -1) { - const void *cookie = io_getcookie(sock); + const void *cookie = fdstate_cookie(sock); if ((uintptr_t)cookie == FLAG_SERVERSOCKET) handle_accept(sock); else @@ -493,7 +494,7 @@ static int64_t ot_try_bind(ot_ip6 ip, uint16_t port) { if (!io_fd(sock)) panic("io_fd"); - io_setcookie(sock, (void *)FLAG_SERVERSOCKET); + fdstate_open(sock, (void *)FLAG_SERVERSOCKET, 0); io_wantread(sock); return sock; } @@ -535,6 +536,8 @@ int main(int argc, char **argv) { int scanon = 1, lbound = 0, sbound = 0; random_init(); + if (fdstate_init() == -1) + panic("fdstate_init"); g_tracker_id = random_uint32(); while (scanon) { -- cgit v1.3