1 files changed, 72 insertions, 46 deletions

diff --git a/opentracker.c b/opentracker.c
index 0f0b314..11da486 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;
   }
@@ -178,12 +177,13 @@ static void handle_read(const int64_t sock, struct ot_workstruct *ws) {
 }
 
 static void handle_write(const int64_t sock) {
-  struct http_data *cookie = fdstate_cookie(sock);
-  int               chunked = 0;
+  struct http_data *cookie     = fdstate_cookie(sock);
+  int               chunked    = 0;
+  int               keep_alive = 0;
 
   if (cookie) {
-    if (cookie->flag & STRUCT_HTTP_FLAG_CHUNKED_IN_TRANSFER)
-      chunked = 1;
+    chunked    = (cookie->flag & STRUCT_HTTP_FLAG_CHUNKED_IN_TRANSFER) != 0;
+    keep_alive = (cookie->flag & STRUCT_HTTP_FLAG_KEEPALIVE) != 0;
 
     if (cookie->batch) {
       if (batch_send(sock, cookie->batch) == BATCH_FAILED) {
@@ -197,8 +197,14 @@ 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);
-  else
+    http_interest_drop(sock, cookie, EVENT_OUT);
+  else if (keep_alive) {
+    /* The reply is out and the client asked to keep the connection, so stop
+       waiting to write and go back to waiting for the next request */
+    http_interest_drop(sock, cookie, EVENT_OUT);
+    http_interest_add(sock, cookie, EVENT_IN);
+    fdstate_set_expiry(sock, g_now_seconds + OT_CLIENT_TIMEOUT);
+  } else
     handle_dead(sock);
 }
 
@@ -210,25 +216,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 +262,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;
+
+    count = event_wait(ready, OT_EVENT_BATCH, OT_CLIENT_TIMEOUT_CHECKINTERVAL * 1000);
+
+    for (i = 0; i < count; ++i) {
+      const void *cookie;
 
-    io_wait();
+      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 +344,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 +374,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);
   }
 }
@@ -375,14 +398,14 @@ void load_state(const char *const state_filename) {
   while (fgets(inbuf, sizeof(inbuf), state_filehandle)) {
     size_t i;
 
-    if (!hex_tohash(infohash, inbuf))
+    if (!hex_to_hash(infohash, inbuf))
       continue;
     i = 2 * sizeof(ot_hash);
 
-    if (inbuf[i++] != ':' || !(consumed = scan_udec(inbuf + i, &base, UINT64_MAX)))
+    if (inbuf[i++] != ':' || !(consumed = scan_uint(inbuf + i, &base, UINT64_MAX)))
       continue;
     i += consumed;
-    if (inbuf[i++] != ':' || !(consumed = scan_udec(inbuf + i, &downcount, UINT64_MAX)))
+    if (inbuf[i++] != ':' || !(consumed = scan_uint(inbuf + i, &downcount, UINT64_MAX)))
       continue;
     add_torrent_from_saved_state(infohash, base, downcount);
   }
@@ -453,6 +476,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 +496,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 +524,7 @@ int main(int argc, char **argv) {
 
   server_mainloop(0);
 
+  event_deinit();
   fdstate_deinit();
   return 0;
 }