1 files changed, 95 insertions, 93 deletions

diff --git a/opentracker.c b/opentracker.c
index 96a265a..eb97b42 100644
--- a/opentracker.c
+++ b/opentracker.c
@@ -21,10 +21,6 @@
 #include <syslog.h>
 #endif
 
-/* Libowfat */
-#include "io.h"
-#include "iob.h"
-
 /* Opentracker */
 #include "ot_accesslist.h"
 #include "ot_config.h"
@@ -35,15 +31,16 @@
 #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"
 #include "ot_udp.h"
 #include "trackerlogic.h"
 
 /* Globals */
-time_t              g_now_seconds;
 uint32_t            g_tracker_id;
-volatile int        g_opentracker_running = 1;
+_Atomic int         g_opentracker_running = 1;
 int                 g_self_pipe[2];
 
 
@@ -118,25 +115,27 @@ 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) {
-    size_t i;
-    for (i = 0; i < cookie->batches; ++i)
-      iob_reset(cookie->batch + i);
+    if (cookie->batch)
+      batch_reset(cookie->batch);
     free(cookie->batch);
     buf_reset(&cookie->request);
     if (cookie->flag & (STRUCT_HTTP_FLAG_WAITINGFORTASK | STRUCT_HTTP_FLAG_CHUNKED_IN_TRANSFER))
       mutex_workqueue_canceltask(sock);
     free(cookie);
   }
-  io_close(sock);
+  /* The generation has to move before the kernel can reissue the number */
+  fdstate_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     = io_getcookie(sock);
-  ssize_t           byte_count = io_tryread(sock, ws->inbuf, G_INBUF_SIZE);
+  struct http_data *cookie     = fdstate_cookie(sock);
+  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;
   }
@@ -177,37 +176,34 @@ 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);
-  size_t            i;
-  int               chunked = 0;
+  struct http_data *cookie     = fdstate_cookie(sock);
+  int               chunked    = 0;
+  int               keep_alive = 0;
 
-  /* Look for the first io_batch still containing bytes to write */
   if (cookie) {
-    if (cookie->flag & STRUCT_HTTP_FLAG_CHUNKED_IN_TRANSFER)
-      chunked = 1;
-
-    for (i = 0; i < cookie->batches; ++i) {
-      if (cookie->batch[i].bytesleft) {
-        int64_t res = iob_send(sock, cookie->batch + i);
-
-        if (res == -3) {
-          handle_dead(sock);
-          return;
-        }
-
-        if (!cookie->batch[i].bytesleft)
-          continue;
+    chunked    = (cookie->flag & STRUCT_HTTP_FLAG_CHUNKED_IN_TRANSFER) != 0;
+    keep_alive = (cookie->flag & STRUCT_HTTP_FLAG_KEEPALIVE) != 0;
 
-        if (res == -1 || res > 0 || i < cookie->batches - 1)
-          return;
+    if (cookie->batch) {
+      if (batch_send(sock, cookie->batch) == BATCH_FAILED) {
+        handle_dead(sock);
+        return;
       }
+      if (batch_pending(cookie->batch))
+        return;
     }
   }
 
-  /* In a chunked transfer after all batches accumulated have been sent, wait for the next one */
+  /* 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, time_future_seconds(OT_CLIENT_TIMEOUT));
+  } else
     handle_dead(sock);
 }
 
@@ -216,37 +212,35 @@ static void handle_accept(const int64_t serversocket) {
   int64_t           sock;
   ot_ip6            ip;
   uint16_t          port;
-  tai6464           t;
 
   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 *)calloc(1, 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;
 
-    io_setcookie(sock, cookie);
-    io_wantread(sock);
+    if (event_add((int)sock, EVENT_IN, fdstate_open(sock, cookie, time_future_seconds(OT_CLIENT_TIMEOUT))) == -1) {
+      fdstate_close(sock);
+      free(cookie);
+      close((int)sock);
+      continue;
+    }
 
     stats_issue_event(EVENT_ACCEPT, FLAG_TCP, (uintptr_t)ip);
-
-    /* That breaks taia encapsulation. But there is no way to take system
-       time this often in FreeBSD and libowfat does not allow to set unix time */
-    taia_uint(&t, 0); /* Clear t */
-    tai_unix(&(t.sec), (g_now_seconds + OT_CLIENT_TIMEOUT));
-    io_timeout(sock, t);
   }
-  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;
+  time_t               next_timeout_check = time_future_seconds(OT_CLIENT_TIMEOUT_CHECKINTERVAL);
   struct iovec        *iovector;
   size_t               iovec_entries;
   int                  is_partial;
@@ -266,32 +260,48 @@ 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 = io_getcookie(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 (time_now_seconds() > next_timeout_check) {
+      int cursor = 0;
 
-    if (g_now_seconds > next_timeout_check) {
-      while ((sock = io_timeouted()) != -1)
+      while ((sock = fdstate_next_expired(time_now_seconds(), &cursor)) != -1)
         handle_dead(sock);
-      next_timeout_check = g_now_seconds + OT_CLIENT_TIMEOUT_CHECKINTERVAL;
+      next_timeout_check = time_future_seconds(OT_CLIENT_TIMEOUT_CHECKINTERVAL);
     }
 
     livesync_ticker();
@@ -332,11 +342,8 @@ static void listen_bind_one(ot_listen *entry) {
       panic("socket_udp_bind");
   }
 
-  if (!io_fd(entry->sock))
-    panic("io_fd");
-
-  io_setcookie(entry->sock, (void *)entry->proto);
-  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);
@@ -365,7 +372,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);
   }
 }
@@ -388,14 +396,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);
   }
@@ -451,19 +459,13 @@ int drop_privileges(const char *const serveruser, const char *const serverdir) {
 }
 
 /* Maintain our copy of the clock. time() on BSDs is very expensive. */
-static void *time_caching_worker(void *args) {
-  (void)args;
-  while (1) {
-    g_now_seconds = time(NULL);
-    sleep(5);
-  }
-  return NULL;
-}
-
 int main(int argc, char **argv) {
-  pthread_t thread_id; /* time cacher */
 
   random_init();
+  if (fdstate_init() == -1)
+    panic("fdstate_init");
+  if (event_init() == -1)
+    panic("event_init");
   config_parse(argc, argv);
 
   listen_bind_all();
@@ -478,19 +480,17 @@ int main(int argc, char **argv) {
   if (drop_privileges(g_serveruser ? g_serveruser : "nobody", g_serverdir) == -1)
     panic("drop_privileges failed, exiting. Last error");
 
-  g_now_seconds = time(NULL);
-  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: ");
-  io_setcookie(g_self_pipe[0], (void *)FLAG_SELFPIPE);
-  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();
@@ -510,5 +510,7 @@ int main(int argc, char **argv) {
 
   server_mainloop(0);
 
+  event_deinit();
+  fdstate_deinit();
   return 0;
 }