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. --- Makefile | 17 +++++++---- opentracker.c | 95 ++++++++++++++++++++++++++++++++++----------------------- ot_accesslist.c | 2 -- ot_clean.c | 3 -- ot_event.c | 20 ++++++------ ot_event.h | 2 +- ot_fdstate.c | 6 ++++ ot_fdstate.h | 4 +++ ot_fullscrape.c | 3 -- ot_http.c | 39 +++++++++++++++-------- ot_http.h | 6 ++++ ot_iovec.c | 2 -- ot_livesync.h | 1 - ot_mutex.c | 9 +++--- ot_socket.c | 6 ++-- ot_socket.h | 1 + ot_stats.c | 3 -- ot_sync.c | 3 -- ot_udp.c | 3 -- tests/t_event.c | 56 +++++++++++++++++----------------- 20 files changed, 159 insertions(+), 122 deletions(-) diff --git a/Makefile b/Makefile index a373633..87225f7 100644 --- a/Makefile +++ b/Makefile @@ -64,8 +64,13 @@ GIT_VERSION=`sh -c 'command -v git >/dev/null && test -d .git && git rev-parse H OPTS_debug=-D_DEBUG -g -ggdb # -pg -fprofile-arcs -ftest-coverage OPTS_production=-O3 -CFLAGS+=-I$(LIBOWFAT_HEADERS) -DGIT_VERSION=$(GIT_VERSION) -Wall -pipe -pthread -Wextra -Wwrite-strings #-ansi -pedantic -LDFLAGS+=-L$(LIBOWFAT_LIBRARY) -lowfat -pthread +CFLAGS+=-DGIT_VERSION=$(GIT_VERSION) -Wall -pipe -pthread -Wextra -Wwrite-strings #-ansi -pedantic +LDFLAGS+=-pthread + +# Only the proxy, which has not been ported since the tracker went dual stack, +# still wants libowfat. +CFLAGS_libowfat=-I$(LIBOWFAT_HEADERS) +LDFLAGS_libowfat=-L$(LIBOWFAT_LIBRARY) -lowfat BINARY =opentracker HEADERS=trackerlogic.h ot_urlencoded.h ot_batch.h ot_buf.h ot_event.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 @@ -84,8 +89,8 @@ all: $(BINARY) $(BINARY).debug CFLAGS_production = $(CFLAGS) $(OPTS_production) $(FEATURES) CFLAGS_debug = $(CFLAGS) $(OPTS_debug) $(FEATURES) -CFLAGS_proxy = $(CFLAGS_production) -DWANT_SYNC_LIVE -CFLAGS_proxy_debug = $(CFLAGS_debug) -DWANT_SYNC_LIVE +CFLAGS_proxy = $(CFLAGS_production) $(CFLAGS_libowfat) -DWANT_SYNC_LIVE +CFLAGS_proxy_debug = $(CFLAGS_debug) $(CFLAGS_libowfat) -DWANT_SYNC_LIVE $(BINARY): $(OBJECTS) $(CC) -o $@ $(OBJECTS) $(LDFLAGS) @@ -93,9 +98,9 @@ $(BINARY): $(OBJECTS) $(BINARY).debug: $(OBJECTS_debug) $(CC) -o $@ $(OBJECTS_debug) $(LDFLAGS) proxy: $(OBJECTS_proxy) - $(CC) -o $@ $(OBJECTS_proxy) $(CFLAGS_production) $(LDFLAGS) + $(CC) -o $@ $(OBJECTS_proxy) $(CFLAGS_production) $(LDFLAGS) $(LDFLAGS_libowfat) proxy.debug: $(OBJECTS_proxy_debug) - $(CC) -o $@ $(OBJECTS_proxy_debug) $(LDFLAGS) + $(CC) -o $@ $(OBJECTS_proxy_debug) $(LDFLAGS) $(LDFLAGS_libowfat) $(OBJECTS): $(HEADERS) $(OBJECTS_debug): $(HEADERS) 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; } diff --git a/ot_accesslist.c b/ot_accesslist.c index 79a11a5..724978e 100644 --- a/ot_accesslist.c +++ b/ot_accesslist.c @@ -21,8 +21,6 @@ #include #endif -/* Libowfat */ - /* Opentracker */ #include "ot_accesslist.h" #include "ot_hex.h" diff --git a/ot_clean.c b/ot_clean.c index 825290d..bf0504c 100644 --- a/ot_clean.c +++ b/ot_clean.c @@ -8,9 +8,6 @@ #include #include -/* Libowfat */ -#include "io.h" - /* Opentracker */ #include "ot_accesslist.h" #include "ot_clean.h" diff --git a/ot_event.c b/ot_event.c index d24227d..ce0feef 100644 --- a/ot_event.c +++ b/ot_event.c @@ -60,7 +60,7 @@ static int event_ctl(int op, int fd, int what, uint32_t generation) { struct epoll_event ev; memset(&ev, 0, sizeof(ev)); - ev.events = (what & EVENT_READ ? (uint32_t)EPOLLIN : 0) | (what & EVENT_WRITE ? (uint32_t)EPOLLOUT : 0); + ev.events = (what & EVENT_IN ? (uint32_t)EPOLLIN : 0) | (what & EVENT_OUT ? (uint32_t)EPOLLOUT : 0); /* The generation rides along inside the kernel, so an event says which connection it belongs to without anybody looking it up */ ev.data.u64 = ((uint64_t)generation << 32) | (uint32_t)fd; @@ -92,8 +92,8 @@ int event_wait(ot_event *out, int max, int timeout_ms) { out[i].generation = (uint32_t)(g_ready[i].data.u64 >> 32); /* A hangup or an error is reported as both, so whichever side the caller was waiting on gets to notice */ - out[i].what = (events & (EPOLLIN | EPOLLHUP | EPOLLERR) ? EVENT_READ : 0) | - (events & (EPOLLOUT | EPOLLHUP | EPOLLERR) ? EVENT_WRITE : 0); + out[i].what = (events & (EPOLLIN | EPOLLHUP | EPOLLERR) ? EVENT_IN : 0) | + (events & (EPOLLOUT | EPOLLHUP | EPOLLERR) ? EVENT_OUT : 0); } return count; } @@ -138,9 +138,9 @@ static int event_change(int fd, int what, uint32_t generation, int deleting) { int succeeded = 0, count, i; EV_SET(&change[0], (uintptr_t)fd, EVFILT_READ, - EV_RECEIPT | ((deleting || !(what & EVENT_READ)) ? EV_DELETE : EV_ADD), 0, 0, udata); + EV_RECEIPT | ((deleting || !(what & EVENT_IN)) ? EV_DELETE : EV_ADD), 0, 0, udata); EV_SET(&change[1], (uintptr_t)fd, EVFILT_WRITE, - EV_RECEIPT | ((deleting || !(what & EVENT_WRITE)) ? EV_DELETE : EV_ADD), 0, 0, udata); + EV_RECEIPT | ((deleting || !(what & EVENT_OUT)) ? EV_DELETE : EV_ADD), 0, 0, udata); if ((count = kevent(g_kqueue, change, 2, result, 2, NULL)) == -1) return -1; @@ -197,8 +197,8 @@ int event_wait(ot_event *out, int max, int timeout_ms) { out[i].fd = (int)(uint32_t)udata; out[i].generation = (uint32_t)(udata >> 32); - out[i].what = ((g_ready[i].filter == EVFILT_READ || eof) ? EVENT_READ : 0) | - ((g_ready[i].filter == EVFILT_WRITE || eof) ? EVENT_WRITE : 0); + out[i].what = ((g_ready[i].filter == EVFILT_READ || eof) ? EVENT_IN : 0) | + ((g_ready[i].filter == EVFILT_WRITE || eof) ? EVENT_OUT : 0); } return count; } @@ -232,7 +232,7 @@ void event_deinit(void) { } static short event_to_poll(int what) { - return (short)((what & EVENT_READ ? POLLIN : 0) | (what & EVENT_WRITE ? POLLOUT : 0)); + return (short)((what & EVENT_IN ? POLLIN : 0) | (what & EVENT_OUT ? POLLOUT : 0)); } static int event_room_for(int fd) { @@ -322,8 +322,8 @@ int event_wait(ot_event *out, int max, int timeout_ms) { continue; out[found].fd = g_pfds[i].fd; out[found].generation = g_generation[i]; - out[found].what = (revents & (POLLIN | POLLHUP | POLLERR | POLLNVAL) ? EVENT_READ : 0) | - (revents & (POLLOUT | POLLHUP | POLLERR | POLLNVAL) ? EVENT_WRITE : 0); + out[found].what = (revents & (POLLIN | POLLHUP | POLLERR | POLLNVAL) ? EVENT_IN : 0) | + (revents & (POLLOUT | POLLHUP | POLLERR | POLLNVAL) ? EVENT_OUT : 0); ++found; } return found; diff --git a/ot_event.h b/ot_event.h index c1627ab..6023134 100644 --- a/ot_event.h +++ b/ot_event.h @@ -14,7 +14,7 @@ the connection it is holding from one for a connection that has since been closed and had its number reused. */ -enum { EVENT_READ = 1, EVENT_WRITE = 2 }; +enum { EVENT_IN = 1, EVENT_OUT = 2 }; typedef struct { int fd; diff --git a/ot_fdstate.c b/ot_fdstate.c index 7f73d3d..537bcb0 100644 --- a/ot_fdstate.c +++ b/ot_fdstate.c @@ -99,6 +99,12 @@ uint32_t fdstate_open(int fd, void *cookie, ot_time expiry) { return FD_GEN(word); } +uint32_t fdstate_generation(int fd) { + ot_fdslot *slot = fdstate_slot(fd); + + return slot ? FD_GEN(atomic_load_explicit(&slot->state, memory_order_relaxed)) : 0; +} + int fdstate_claim(int fd, uint32_t generation) { ot_fdslot *slot = fdstate_slot(fd); uint64_t word; diff --git a/ot_fdstate.h b/ot_fdstate.h index 88f685d..c09f018 100644 --- a/ot_fdstate.h +++ b/ot_fdstate.h @@ -31,6 +31,10 @@ size_t fdstate_slots(void); generation to quote when asking for it back. expiry of 0 never expires. */ uint32_t fdstate_open(int fd, void *cookie, ot_time expiry); +/* Which generation a descriptor currently carries, to check an event that + was handed out earlier against the connection living there now. */ +uint32_t fdstate_generation(int fd); + /* Ask for exclusive use. Returns 0 if the descriptor has been recycled since generation was handed out, or if somebody else is using it. */ int fdstate_claim(int fd, uint32_t generation); diff --git a/ot_fullscrape.c b/ot_fullscrape.c index 0dee5c2..2ba3b24 100644 --- a/ot_fullscrape.c +++ b/ot_fullscrape.c @@ -17,9 +17,6 @@ #include #endif -/* Libowfat */ -#include "io.h" - /* Opentracker */ #include "ot_hex.h" #include "ot_fullscrape.h" diff --git a/ot_http.c b/ot_http.c index 3c135cb..5e10d59 100644 --- a/ot_http.c +++ b/ot_http.c @@ -14,9 +14,6 @@ #include #include -/* Libowfat */ -#include "io.h" - /* Opentracker */ #include "ot_accesslist.h" #include "ot_fullscrape.h" @@ -25,6 +22,7 @@ #include "ot_ip.h" #include "ot_mutex.h" #include "ot_scan.h" +#include "ot_event.h" #include "ot_fdstate.h" #include "ot_stats.h" #include "ot_urlencoded.h" @@ -43,7 +41,7 @@ static void http_senddata(const int64_t sock, struct ot_workstruct *ws) { ssize_t written_size; if (!cookie) { - io_close(sock); + event_del((int)sock), close((int)sock); return; } @@ -67,7 +65,7 @@ static void http_senddata(const int64_t sock, struct ot_workstruct *ws) { if ((written_size < 0) || ((written_size == ws->reply_size) && !ws->keep_alive)) { buf_reset(&cookie->request); free(cookie); - io_close(sock); + event_del((int)sock), close((int)sock); return; } @@ -77,7 +75,7 @@ static void http_senddata(const int64_t sock, struct ot_workstruct *ws) { if (!(outbuf = malloc(ws->reply_size - written_size))) { buf_reset(&cookie->request); free(cookie); - io_close(sock); + event_del((int)sock), close((int)sock); return; } @@ -88,7 +86,7 @@ static void http_senddata(const int64_t sock, struct ot_workstruct *ws) { free(outbuf); buf_reset(&cookie->request); free(cookie); - io_close(sock); + event_del((int)sock), close((int)sock); return; } } @@ -98,9 +96,9 @@ 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) { fdstate_set_expiry(sock, 0); - io_dontwantread(sock); + http_interest_drop(sock, cookie, EVENT_IN); } - io_wantwrite(sock); + http_interest_add(sock, cookie, EVENT_OUT); } } @@ -113,6 +111,21 @@ static void http_senddata(const int64_t sock, struct ot_workstruct *ws) { #define HTTPERROR_403_IP return http_issue_error(sock, ws, CODE_HTTPERROR_403_IP) #define HTTPERROR_404 return http_issue_error(sock, ws, CODE_HTTPERROR_404) #define HTTPERROR_500 return http_issue_error(sock, ws, CODE_HTTPERROR_500) +static void http_interest(const int64_t sock, struct http_data *cookie, int what) { + if (cookie->interest == what) + return; + cookie->interest = what; + event_modify((int)sock, what, fdstate_generation((int)sock)); +} + +void http_interest_add(const int64_t sock, struct http_data *cookie, int what) { + http_interest(sock, cookie, cookie->interest | what); +} + +void http_interest_drop(const int64_t sock, struct http_data *cookie, int what) { + http_interest(sock, cookie, cookie->interest & ~what); +} + ssize_t http_issue_error(const int64_t sock, struct ot_workstruct *ws, int code) { const char *error_code[] = {"302 Found", "400 Invalid Request", "400 Invalid Request", "400 Invalid Request", "402 Payment Required", "403 Not Modest", "403 Access Denied", "404 Not Found", "500 Internal Server Error"}; @@ -157,7 +170,7 @@ ssize_t http_sendiovecdata(const int64_t sock, struct ot_workstruct *ws, size_t /* If we came here, wait for the answer is over */ if (cookie->flag & STRUCT_HTTP_FLAG_WAITINGFORTASK) { - io_dontwantread(sock); + http_interest_drop(sock, cookie, EVENT_IN); cookie->flag &= ~STRUCT_HTTP_FLAG_WAITINGFORTASK; } @@ -208,7 +221,7 @@ ssize_t http_sendiovecdata(const int64_t sock, struct ot_workstruct *ws, size_t } fdstate_set_expiry(sock, g_now_seconds + OT_CLIENT_TIMEOUT_SEND); - io_wantwrite(sock); + http_interest_add(sock, cookie, EVENT_OUT); return 0; } @@ -303,7 +316,7 @@ static ssize_t http_handle_stats(const int64_t sock, struct ot_workstruct *ws, c /* Clients waiting for us should not easily timeout */ fdstate_set_expiry(sock, 0); fullscrape_deliver(sock, format); - io_dontwantread(sock); + http_interest_drop(sock, cookie, EVENT_IN); return ws->reply_size = -2; } #endif @@ -396,7 +409,7 @@ static ssize_t http_handle_fullscrape(const int64_t sock, struct ot_workstruct * /* Clients waiting for us should not easily timeout */ fdstate_set_expiry(sock, 0); fullscrape_deliver(sock, format); - io_dontwantread(sock); + http_interest_drop(sock, cookie, EVENT_IN); return ws->reply_size = -2; } #endif diff --git a/ot_http.h b/ot_http.h index 003b545..1631220 100644 --- a/ot_http.h +++ b/ot_http.h @@ -11,6 +11,7 @@ #include #include "ot_batch.h" +#include "ot_event.h" #include "ot_buf.h" #include "trackerlogic.h" @@ -25,11 +26,16 @@ typedef enum { struct http_data { ot_buf request; + int interest; /* EVENT_IN and EVENT_OUT as asked for */ ot_batch *batch; ot_ip6 ip; STRUCT_HTTP_FLAG flag; }; +/* Add to or drop from what this connection is waiting to be told about */ +void http_interest_add(const int64_t s, struct http_data *cookie, int what); +void http_interest_drop(const int64_t s, struct http_data *cookie, int what); + ssize_t http_handle_request(const int64_t s, struct ot_workstruct *ws); ssize_t http_sendiovecdata(const int64_t s, struct ot_workstruct *ws, size_t iovec_entries, struct iovec *iovector, int is_partial); ssize_t http_issue_error(const int64_t s, struct ot_workstruct *ws, int code); diff --git a/ot_iovec.c b/ot_iovec.c index 2ee455c..7e5903f 100644 --- a/ot_iovec.c +++ b/ot_iovec.c @@ -7,8 +7,6 @@ #include #include -/* Libowfat */ - /* Opentracker */ #include "ot_iovec.h" diff --git a/ot_livesync.h b/ot_livesync.h index 5b9fa0e..4725023 100644 --- a/ot_livesync.h +++ b/ot_livesync.h @@ -6,7 +6,6 @@ #ifndef OT_LIVESYNC_H__ #define OT_LIVESYNC_H__ -#include "io.h" #include "trackerlogic.h" /* diff --git a/ot_mutex.c b/ot_mutex.c index 2a8cbf5..37d9962 100644 --- a/ot_mutex.c +++ b/ot_mutex.c @@ -8,11 +8,10 @@ #include #include #include +#include #include #include -/* Libowfat */ -#include "io.h" /* Opentracker */ #include "ot_iovec.h" @@ -193,7 +192,8 @@ int mutex_workqueue_pushresult(ot_taskid taskid, size_t iovec_entries, struct io /* Release lock */ pthread_mutex_unlock(&tasklist_mutex); - io_trywrite(g_self_pipe[1], &byte, 1); + /* A full pipe has already done this nudge's job, so a failure is fine */ + (void)!write(g_self_pipe[1], &byte, 1); /* Indicate whether the worker has to throw away results */ return task ? 0 : -1; @@ -221,7 +221,8 @@ int mutex_workqueue_pushchunked(ot_taskid taskid, struct iovec *iovec) { /* Release lock */ pthread_mutex_unlock(&tasklist_mutex); - io_trywrite(g_self_pipe[1], &byte, 1); + /* A full pipe has already done this nudge's job, so a failure is fine */ + (void)!write(g_self_pipe[1], &byte, 1); /* Indicate whether the worker has to throw away results */ return task ? 0 : -1; diff --git a/ot_socket.c b/ot_socket.c index 84020ca..576087f 100644 --- a/ot_socket.c +++ b/ot_socket.c @@ -40,14 +40,16 @@ static void ip_from_sockaddr(ot_ip6 ip, uint16_t *port, const struct sockaddr_st } } -static int set_nonblocking(int s) { +int socket_set_blocking(int s, int blocking) { int flags = fcntl(s, F_GETFL, 0); if (flags == -1) return -1; - return fcntl(s, F_SETFL, flags | O_NONBLOCK); + return fcntl(s, F_SETFL, blocking ? (flags & ~O_NONBLOCK) : (flags | O_NONBLOCK)); } +static int set_nonblocking(int s) { return socket_set_blocking(s, 0); } + /* Close without letting the failure we are about to report be overwritten */ static int close_and_fail(int s) { int saved = errno; diff --git a/ot_socket.h b/ot_socket.h index 21d04b1..aa4e2d2 100644 --- a/ot_socket.h +++ b/ot_socket.h @@ -27,6 +27,7 @@ int socket_udp_bind(const ot_ip6 ip, uint16_t port, socket_sharing sharing); int socket_tcp_connect(const ot_ip6 srcip, uint16_t srcport, const ot_ip6 dstip, uint16_t dstport); int socket_is_connected(int s); int socket_accept(int s, ot_ip6 ip, uint16_t *port); +int socket_set_blocking(int s, int blocking); ssize_t socket_recv(int s, void *buf, size_t len, ot_ip6 ip, uint16_t *port); ssize_t socket_send(int s, const void *buf, size_t len, const ot_ip6 ip, uint16_t port); diff --git a/ot_stats.c b/ot_stats.c index ceeb89d..bee1f93 100644 --- a/ot_stats.c +++ b/ot_stats.c @@ -16,9 +16,6 @@ #include #endif -/* Libowfat */ -#include "io.h" - /* Opentracker */ #include "ot_accesslist.h" #include "ot_hex.h" diff --git a/ot_sync.c b/ot_sync.c index 1a6bebc..f958a20 100644 --- a/ot_sync.c +++ b/ot_sync.c @@ -10,9 +10,6 @@ #include #include -/* Libowfat */ -#include "io.h" - /* Opentracker */ #include "ot_iovec.h" #include "ot_mutex.h" diff --git a/ot_udp.c b/ot_udp.c index 7f6517f..5bf9d52 100644 --- a/ot_udp.c +++ b/ot_udp.c @@ -10,9 +10,6 @@ #include #include -/* Libowfat */ -#include "io.h" - /* Opentracker */ #include "ot_ip.h" #include "ot_random.h" diff --git a/tests/t_event.c b/tests/t_event.c index 306f8cd..5509f19 100644 --- a/tests/t_event.c +++ b/tests/t_event.c @@ -59,18 +59,18 @@ static void case_readable(void) { test_case("a descriptor with something to read is reported, with its generation"); pair_open(); CHECK(sv[0] != -1); - CHECK_INT(event_add(sv[0], EVENT_READ, 4711), 0); + CHECK_INT(event_add(sv[0], EVENT_IN, 4711), 0); CHECK_INT(write(sv[1], "x", 1), 1); - CHECK_INT(wait_for(sv[0], EVENT_READ, 1000, &generation), 1); + CHECK_INT(wait_for(sv[0], EVENT_IN, 1000, &generation), 1); CHECK_INT(generation, 4711); test_case("and keeps being reported until it is read"); - CHECK_INT(wait_for(sv[0], EVENT_READ, 1000, NULL), 1); + CHECK_INT(wait_for(sv[0], EVENT_IN, 1000, NULL), 1); { char scratch[4]; CHECK_INT(read(sv[0], scratch, sizeof(scratch)), 1); } - CHECK_INT(wait_for(sv[0], EVENT_READ, 50, NULL), 0); + CHECK_INT(wait_for(sv[0], EVENT_IN, 50, NULL), 0); CHECK_INT(event_del(sv[0]), 0); pair_close(); @@ -80,12 +80,12 @@ static void case_writable(void) { test_case("a descriptor with room to write is reported"); pair_open(); CHECK(sv[0] != -1); - CHECK_INT(event_add(sv[0], EVENT_WRITE, 1), 0); - CHECK_INT(wait_for(sv[0], EVENT_WRITE, 1000, NULL), 1); + CHECK_INT(event_add(sv[0], EVENT_OUT, 1), 0); + CHECK_INT(wait_for(sv[0], EVENT_OUT, 1000, NULL), 1); test_case("and is not, once we say we no longer care"); - CHECK_INT(event_modify(sv[0], EVENT_READ, 1), 0); - CHECK_INT(wait_for(sv[0], EVENT_WRITE, 50, NULL), 0); + CHECK_INT(event_modify(sv[0], EVENT_IN, 1), 0); + CHECK_INT(wait_for(sv[0], EVENT_OUT, 50, NULL), 0); CHECK_INT(event_del(sv[0]), 0); pair_close(); @@ -95,13 +95,13 @@ static void case_interest_is_separate(void) { test_case("asking only about writing does not report reading"); pair_open(); CHECK(sv[0] != -1); - CHECK_INT(event_add(sv[0], EVENT_WRITE, 1), 0); + CHECK_INT(event_add(sv[0], EVENT_OUT, 1), 0); CHECK_INT(write(sv[1], "x", 1), 1); - CHECK_INT(wait_for(sv[0], EVENT_READ, 50, NULL), 0); + CHECK_INT(wait_for(sv[0], EVENT_IN, 50, NULL), 0); test_case("until we ask about it"); - CHECK_INT(event_modify(sv[0], EVENT_READ | EVENT_WRITE, 1), 0); - CHECK_INT(wait_for(sv[0], EVENT_READ, 1000, NULL), 1); + CHECK_INT(event_modify(sv[0], EVENT_IN | EVENT_OUT, 1), 0); + CHECK_INT(wait_for(sv[0], EVENT_IN, 1000, NULL), 1); CHECK_INT(event_del(sv[0]), 0); pair_close(); @@ -111,11 +111,11 @@ static void case_removed_is_silent(void) { test_case("a removed descriptor is not reported again"); pair_open(); CHECK(sv[0] != -1); - CHECK_INT(event_add(sv[0], EVENT_READ, 1), 0); + CHECK_INT(event_add(sv[0], EVENT_IN, 1), 0); CHECK_INT(write(sv[1], "x", 1), 1); - CHECK_INT(wait_for(sv[0], EVENT_READ, 1000, NULL), 1); + CHECK_INT(wait_for(sv[0], EVENT_IN, 1000, NULL), 1); CHECK_INT(event_del(sv[0]), 0); - CHECK_INT(wait_for(sv[0], EVENT_READ, 50, NULL), 0); + CHECK_INT(wait_for(sv[0], EVENT_IN, 50, NULL), 0); pair_close(); } @@ -127,21 +127,21 @@ static void case_generations_do_not_mix(void) { pair_open(); CHECK(sv[0] != -1); CHECK_INT(socketpair(AF_UNIX, SOCK_STREAM, 0, other), 0); - CHECK_INT(event_add(sv[0], EVENT_READ, 100), 0); - CHECK_INT(event_add(other[0], EVENT_READ, 200), 0); + CHECK_INT(event_add(sv[0], EVENT_IN, 100), 0); + CHECK_INT(event_add(other[0], EVENT_IN, 200), 0); CHECK_INT(write(sv[1], "x", 1), 1); CHECK_INT(write(other[1], "y", 1), 1); - CHECK_INT(wait_for(sv[0], EVENT_READ, 1000, &generation), 1); + CHECK_INT(wait_for(sv[0], EVENT_IN, 1000, &generation), 1); CHECK_INT(generation, 100); - CHECK_INT(wait_for(other[0], EVENT_READ, 1000, &generation), 1); + CHECK_INT(wait_for(other[0], EVENT_IN, 1000, &generation), 1); CHECK_INT(generation, 200); test_case("and re-registering one changes only its own"); - CHECK_INT(event_modify(sv[0], EVENT_READ, 101), 0); - CHECK_INT(wait_for(sv[0], EVENT_READ, 1000, &generation), 1); + CHECK_INT(event_modify(sv[0], EVENT_IN, 101), 0); + CHECK_INT(wait_for(sv[0], EVENT_IN, 1000, &generation), 1); CHECK_INT(generation, 101); - CHECK_INT(wait_for(other[0], EVENT_READ, 1000, &generation), 1); + CHECK_INT(wait_for(other[0], EVENT_IN, 1000, &generation), 1); CHECK_INT(generation, 200); event_del(sv[0]); @@ -162,23 +162,23 @@ static void case_delete_from_the_middle(void) { test_case("deleting one of three leaves the others reported correctly"); for (i = 0; i < 3; ++i) { CHECK_INT(socketpair(AF_UNIX, SOCK_STREAM, 0, pair[i]), 0); - CHECK_INT(event_add(pair[i][0], EVENT_READ, (uint32_t)(10 + i)), 0); + CHECK_INT(event_add(pair[i][0], EVENT_IN, (uint32_t)(10 + i)), 0); } CHECK_INT(event_del(pair[1][0]), 0); for (i = 0; i < 3; i += 2) { CHECK_INT(write(pair[i][1], "x", 1), 1); - CHECK_INT(wait_for(pair[i][0], EVENT_READ, 1000, &generation), 1); + CHECK_INT(wait_for(pair[i][0], EVENT_IN, 1000, &generation), 1); CHECK_INT(generation, 10 + i); } test_case("and the survivors can still be changed and removed"); - CHECK_INT(event_modify(pair[2][0], EVENT_READ, 99), 0); - CHECK_INT(wait_for(pair[2][0], EVENT_READ, 1000, &generation), 1); + CHECK_INT(event_modify(pair[2][0], EVENT_IN, 99), 0); + CHECK_INT(wait_for(pair[2][0], EVENT_IN, 1000, &generation), 1); CHECK_INT(generation, 99); CHECK_INT(event_del(pair[0][0]), 0); CHECK_INT(event_del(pair[2][0]), 0); - CHECK_INT(wait_for(pair[2][0], EVENT_READ, 50, NULL), 0); + CHECK_INT(wait_for(pair[2][0], EVENT_IN, 50, NULL), 0); for (i = 0; i < 3; ++i) close(pair[i][0]), close(pair[i][1]); @@ -201,7 +201,7 @@ static void case_timeout(void) { static void case_unknown_descriptor(void) { test_case("changing or removing one that was never added is refused"); - CHECK_INT(event_modify(4096, EVENT_READ, 1), -1); + CHECK_INT(event_modify(4096, EVENT_IN, 1), -1); CHECK_INT(event_del(4096), -1); } -- cgit v1.3