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 <syslog.h>
 #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 <errno.h>
 #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 <string.h>
 #include <unistd.h>
 
-/* 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 <zstd.h>
 #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 <sys/types.h>
 #include <unistd.h>
 
-/* 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 <sys/uio.h>
 
 #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 <stdlib.h>
 #include <sys/uio.h>
 
-/* 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 <pthread.h>
 #include <stdio.h>
 #include <stdlib.h>
+#include <unistd.h>
 #include <string.h>
 #include <sys/uio.h>
 
-/* 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 <syslog.h>
 #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 <sys/types.h>
 #include <sys/uio.h>
 
-/* 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 <stdlib.h>
 #include <string.h>
 
-/* 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);
 }