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.

1 files changed, 57 insertions, 38 deletions

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;
 }