7 files changed, 69 insertions, 70 deletions

diff --git a/Makefile b/Makefile
index 3bf254a..87c20e2 100644
--- a/Makefile
+++ b/Makefile
@@ -70,7 +70,7 @@ LDFLAGS+=-L$(LIBOWFAT_LIBRARY) -lowfat -pthread
 BINARY =opentracker
 HEADERS=trackerlogic.h scan_urlencoded_query.h ot_batch.h ot_buf.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
 SOURCES=opentracker.c trackerlogic.c scan_urlencoded_query.c ot_batch.c ot_buf.c ot_fdstate.c ot_hex.c ot_ip.c ot_config.c ot_scan.c ot_socket.c ot_mutex.c ot_random.c ot_stats.c ot_vector.c ot_clean.c ot_udp.c ot_iovec.c ot_fullscrape.c ot_accesslist.c ot_http.c ot_livesync.c ot_rijndael.c
-SOURCES_proxy=proxy.c ot_batch.c ot_vector.c ot_mutex.c ot_random.c ot_scan.c ot_ip.c ot_socket.c
+SOURCES_proxy=proxy.c ot_batch.c ot_fdstate.c ot_vector.c ot_mutex.c ot_random.c ot_scan.c ot_ip.c ot_socket.c
 
 OBJECTS = $(SOURCES:%.c=%.o)
 OBJECTS_debug = $(SOURCES:%.c=%.debug.o)
diff --git a/opentracker.c b/opentracker.c
index 60bdceb..0f0b314 100644
--- a/opentracker.c
+++ b/opentracker.c
@@ -34,8 +34,8 @@
 #include "ot_mutex.h"
 #include "ot_random.h"
 #include "ot_scan.h"
+#include "ot_fdstate.h"
 #include "ot_socket.h"
-#include "ot_timeout.h"
 #include "ot_stats.h"
 #include "ot_udp.h"
 #include "trackerlogic.h"
@@ -118,7 +118,7 @@ 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) {
     if (cookie->batch)
       batch_reset(cookie->batch);
@@ -128,11 +128,13 @@ static void handle_dead(const int64_t sock) {
       mutex_workqueue_canceltask(sock);
     free(cookie);
   }
+  /* The generation has to move before the kernel can reissue the number */
+  fdstate_close(sock);
   io_close(sock);
 }
 
 static void handle_read(const int64_t sock, struct ot_workstruct *ws) {
-  struct http_data *cookie     = io_getcookie(sock);
+  struct http_data *cookie     = fdstate_cookie(sock);
   ssize_t           byte_count = io_tryread(sock, ws->inbuf, G_INBUF_SIZE);
 
   if (byte_count == 0 || byte_count == -3) {
@@ -176,7 +178,7 @@ 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);
+  struct http_data *cookie = fdstate_cookie(sock);
   int               chunked = 0;
 
   if (cookie) {
@@ -218,12 +220,11 @@ static void handle_accept(const int64_t serversocket) {
     memset(cookie, 0, sizeof(struct http_data));
     memcpy(cookie->ip, ip, sizeof(ot_ip6));
 
-    io_setcookie(sock, cookie);
+    fdstate_open(sock, cookie, g_now_seconds + OT_CLIENT_TIMEOUT);
     io_wantread(sock);
 
     stats_issue_event(EVENT_ACCEPT, FLAG_TCP, (uintptr_t)ip);
 
-    timeout_in(sock, OT_CLIENT_TIMEOUT);
   }
   io_eagain(serversocket);
 }
@@ -255,7 +256,7 @@ static void *server_mainloop(void *args) {
     io_wait();
 
     while ((sock = io_canread()) != -1) {
-      const void *cookie = io_getcookie(sock);
+      const void *cookie = fdstate_cookie(sock);
       if ((intptr_t)cookie == FLAG_TCP)
         handle_accept(sock);
       else if ((intptr_t)cookie == FLAG_UDP)
@@ -273,7 +274,9 @@ static void *server_mainloop(void *args) {
       handle_write(sock);
 
     if (g_now_seconds > next_timeout_check) {
-      while ((sock = io_timeouted()) != -1)
+      int cursor = 0;
+
+      while ((sock = fdstate_next_expired(g_now_seconds, &cursor)) != -1)
         handle_dead(sock);
       next_timeout_check = g_now_seconds + OT_CLIENT_TIMEOUT_CHECKINTERVAL;
     }
@@ -319,7 +322,7 @@ static void listen_bind_one(ot_listen *entry) {
   if (!io_fd(entry->sock))
     panic("io_fd");
 
-  io_setcookie(entry->sock, (void *)entry->proto);
+  fdstate_open(entry->sock, (void *)entry->proto, 0);
   io_wantread(entry->sock);
 
 #ifdef _DEBUG
@@ -448,6 +451,8 @@ int main(int argc, char **argv) {
   pthread_t thread_id; /* time cacher */
 
   random_init();
+  if (fdstate_init() == -1)
+    panic("fdstate_init");
   config_parse(argc, argv);
 
   listen_bind_all();
@@ -473,7 +478,7 @@ int main(int argc, char **argv) {
     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);
+  fdstate_open(g_self_pipe[0], (void *)FLAG_SELFPIPE, 0);
   io_wantread(g_self_pipe[0]);
 
   /* Init all sub systems. This call may fail with an exit() */
@@ -494,5 +499,6 @@ int main(int argc, char **argv) {
 
   server_mainloop(0);
 
+  fdstate_deinit();
   return 0;
 }
diff --git a/ot_fdstate.c b/ot_fdstate.c
index 6ddbea3..7f73d3d 100644
--- a/ot_fdstate.c
+++ b/ot_fdstate.c
@@ -34,8 +34,11 @@ typedef struct {
    range descriptor is rejected */
 #define FD_SLOTS_MAX (1u << 20)
 
-static ot_fdslot *g_slots;
-static size_t     g_slot_count;
+static ot_fdslot      *g_slots;
+static size_t          g_slot_count;
+/* Descriptors are handed out lowest first, so a sweep only has to walk as
+   far as the highest one ever seen rather than the whole table. */
+static _Atomic size_t  g_slot_highest;
 
 int fdstate_init(void) {
   struct rlimit limit;
@@ -51,6 +54,7 @@ int fdstate_init(void) {
   }
 
   g_slot_count = count;
+  atomic_store(&g_slot_highest, 0);
   return 0;
 }
 
@@ -59,6 +63,7 @@ void fdstate_deinit(void) {
     munmap(g_slots, g_slot_count * sizeof(ot_fdslot));
   g_slots      = NULL;
   g_slot_count = 0;
+  atomic_store(&g_slot_highest, 0);
 }
 
 size_t fdstate_slots(void) { return g_slot_count; }
@@ -76,6 +81,15 @@ uint32_t fdstate_open(int fd, void *cookie, ot_time expiry) {
   if (!slot)
     return 0;
 
+  {
+    size_t highest = atomic_load_explicit(&g_slot_highest, memory_order_relaxed);
+
+    while ((size_t)fd > highest &&
+           !atomic_compare_exchange_weak_explicit(&g_slot_highest, &highest, (size_t)fd, memory_order_relaxed,
+                                                  memory_order_relaxed))
+      ;
+  }
+
   word = atomic_load_explicit(&slot->state, memory_order_relaxed);
   atomic_store_explicit(&slot->cookie, cookie, memory_order_relaxed);
   atomic_store_explicit(&slot->expiry, expiry, memory_order_relaxed);
@@ -150,6 +164,16 @@ void fdstate_close(int fd) {
   atomic_store_explicit(&slot->state, FD_WORD(FD_GEN(word) + 1, FD_FREE), memory_order_release);
 }
 
+int fdstate_next_expired(ot_time now, int *cursor) {
+  size_t highest = atomic_load_explicit(&g_slot_highest, memory_order_relaxed);
+
+  for (; (size_t)*cursor <= highest; ++*cursor)
+    if (fdstate_claim_expired(*cursor, now))
+      return (*cursor)++;
+
+  return -1;
+}
+
 void *fdstate_cookie(int fd) {
   ot_fdslot *slot = fdstate_slot(fd);
 
diff --git a/ot_fdstate.h b/ot_fdstate.h
index 2693a3b..88f685d 100644
--- a/ot_fdstate.h
+++ b/ot_fdstate.h
@@ -45,6 +45,11 @@ int      fdstate_claim_expired(int fd, ot_time now);
    quoting the old generation are still in flight. */
 void     fdstate_close(int fd);
 
+/* Walks the table for descriptors that have expired, handing back one at a
+   time until it returns -1. The cursor belongs to the caller, so several
+   sweepers can run without sharing anything. */
+int      fdstate_next_expired(ot_time now, int *cursor);
+
 /* Both only valid while the caller holds the descriptor. */
 void    *fdstate_cookie(int fd);
 void     fdstate_set_expiry(int fd, ot_time expiry);
diff --git a/ot_http.c b/ot_http.c
index 331a457..6bf343c 100644
--- a/ot_http.c
+++ b/ot_http.c
@@ -25,8 +25,8 @@
 #include "ot_ip.h"
 #include "ot_mutex.h"
 #include "ot_scan.h"
+#include "ot_fdstate.h"
 #include "ot_stats.h"
-#include "ot_timeout.h"
 #include "scan_urlencoded_query.h"
 #include "trackerlogic.h"
 
@@ -39,7 +39,7 @@ ssize_t      g_stats_path_len;
 enum { SUCCESS_HTTP_HEADER_LENGTH = 80, SUCCESS_HTTP_SIZE_OFF = 17 };
 
 static void http_senddata(const int64_t sock, struct ot_workstruct *ws) {
-  struct http_data *cookie = io_getcookie(sock);
+  struct http_data *cookie = fdstate_cookie(sock);
   ssize_t           written_size;
 
   if (!cookie) {
@@ -97,7 +97,7 @@ 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) {
-      timeout_never(sock);
+      fdstate_set_expiry(sock, 0);
       io_dontwantread(sock);
     }
     io_wantwrite(sock);
@@ -134,7 +134,7 @@ ssize_t http_issue_error(const int64_t sock, struct ot_workstruct *ws, int code)
 }
 
 ssize_t http_sendiovecdata(const int64_t sock, struct ot_workstruct *ws, size_t iovec_entries, struct iovec *iovector, int is_partial) {
-  struct http_data *cookie = io_getcookie(sock);
+  struct http_data *cookie = fdstate_cookie(sock);
   char             *header;
   const char       *encoding = "";
   size_t            header_size, size = iovec_length(&iovec_entries, (const struct iovec **)&iovector);
@@ -207,7 +207,7 @@ ssize_t http_sendiovecdata(const int64_t sock, struct ot_workstruct *ws, size_t
     cookie->flag &= ~STRUCT_HTTP_FLAG_CHUNKED_IN_TRANSFER;
   }
 
-  timeout_in(sock, OT_CLIENT_TIMEOUT_SEND);
+  fdstate_set_expiry(sock, g_now_seconds + OT_CLIENT_TIMEOUT_SEND);
   io_wantwrite(sock);
   return 0;
 }
@@ -243,7 +243,7 @@ static ssize_t http_handle_stats(const int64_t sock, struct ot_workstruct *ws, c
                                                 {"url", TASK_FULLSCRAPE_TPB_URLENCODED},  {"txt", TASK_FULLSCRAPE_TPB_ASCII},
                                                 {"txtp", TASK_FULLSCRAPE_TPB_ASCII_PLUS}, {NULL, -3}};
 
-  struct http_data *cookie = io_getcookie(sock);
+  struct http_data *cookie = fdstate_cookie(sock);
   int               mode = TASK_STATS_PEERS, scanon = 1, format = 0;
 
   /* A socket without one is a dead socket, which is what http_sendiovecdata
@@ -301,7 +301,7 @@ static ssize_t http_handle_stats(const int64_t sock, struct ot_workstruct *ws, c
     cookie->flag |= STRUCT_HTTP_FLAG_WAITINGFORTASK | STRUCT_HTTP_FLAG_CHUNKED;
 
     /* Clients waiting for us should not easily timeout */
-    timeout_never(sock);
+    fdstate_set_expiry(sock, 0);
     fullscrape_deliver(sock, format);
     io_dontwantread(sock);
     return ws->reply_size = -2;
@@ -314,7 +314,7 @@ static ssize_t http_handle_stats(const int64_t sock, struct ot_workstruct *ws, c
     cookie->flag |= STRUCT_HTTP_FLAG_WAITINGFORTASK;
 
     /* Complex stats also include expensive memory debugging tools */
-    timeout_never(sock);
+    fdstate_set_expiry(sock, 0);
     stats_deliver(sock, mode);
     return ws->reply_size = -2;
   }
@@ -334,7 +334,7 @@ typedef struct {
 
 #ifdef WANT_FULLSCRAPE
 static ssize_t http_handle_fullscrape(const int64_t sock, struct ot_workstruct *ws) {
-  struct http_data *cookie = io_getcookie(sock);
+  struct http_data *cookie = fdstate_cookie(sock);
   int               format = 0;
 
 #ifdef WANT_MODEST_FULLSCRAPES
@@ -394,7 +394,7 @@ static ssize_t http_handle_fullscrape(const int64_t sock, struct ot_workstruct *
   /* Pass this task to the worker thread */
   cookie->flag |= STRUCT_HTTP_FLAG_WAITINGFORTASK | STRUCT_HTTP_FLAG_CHUNKED;
   /* Clients waiting for us should not easily timeout */
-  timeout_never(sock);
+  fdstate_set_expiry(sock, 0);
   fullscrape_deliver(sock, format);
   io_dontwantread(sock);
   return ws->reply_size = -2;
@@ -486,7 +486,7 @@ static ssize_t     http_handle_announce(const int64_t sock, struct ot_workstruct
   unsigned short    port = 0;
   char             *write_ptr;
   ssize_t           len;
-  struct http_data *cookie = io_getcookie(sock);
+  struct http_data *cookie = fdstate_cookie(sock);
 
   /* This is to hack around stupid clients that send "announce ?info_hash" */
   if (read_ptr[-1] != '?') {
@@ -663,7 +663,7 @@ ssize_t http_handle_request(const int64_t sock, struct ot_workstruct *ws) {
   char   *read_ptr = ws->request, *write_ptr;
 
 #ifdef WANT_FULLLOG_NETWORKS
-  struct http_data *cookie = io_getcookie(sock);
+  struct http_data *cookie = fdstate_cookie(sock);
   if (loglist_check_address(cookie->ip)) {
     ot_log *log = malloc(sizeof(ot_log));
     if (log) {
diff --git a/ot_timeout.h b/ot_timeout.h
deleted file mode 100644
index 1027c2e..0000000
--- a/ot_timeout.h
+++ /dev/null
@@ -1,39 +0,0 @@
-/* This software was written by Dirk Engling <erdgeist@erdgeist.org>
-   It is considered beerware. Prost. Skol. Cheers or whatever.
-
-   $id$ */
-
-#ifndef OT_TIMEOUT_H__
-#define OT_TIMEOUT_H__
-
-#include <stdint.h>
-#include <string.h>
-
-/* Libowfat */
-#include "io.h"
-
-#include "trackerlogic.h"
-
-/* io_timeout wants an absolute TAI64 deadline and io_timeouted skips any
-   socket whose deadline is zero, so that is how a socket is told to wait
-   indefinitely.
-
-   The deadline is taken from the cached clock rather than from
-   gettimeofday, which is too expensive to call at this rate on FreeBSD. */
-
-static inline void timeout_never(int64_t sock) {
-  tai6464 t;
-
-  memset(&t, 0, sizeof(t));
-  io_timeout(sock, t);
-}
-
-static inline void timeout_in(int64_t sock, unsigned int seconds) {
-  tai6464 t;
-
-  memset(&t, 0, sizeof(t));
-  tai_unix(&t.sec, g_now_seconds + seconds);
-  io_timeout(sock, t);
-}
-
-#endif
diff --git a/proxy.c b/proxy.c
index 63b4c4f..9486e75 100644
--- a/proxy.c
+++ b/proxy.c
@@ -19,6 +19,7 @@
 
 /* Opentracker */
 #include "ot_batch.h"
+#include "ot_fdstate.h"
 #include "ot_ip.h"
 #include "ot_mutex.h"
 #include "ot_random.h"
@@ -267,7 +268,7 @@ static void handle_reconnects(void) {
         continue; /* No socket for you */
       io_fd(newfd);
       io_wantwrite(newfd); /* So we will be informed when it is connected */
-      io_setcookie(newfd, g_connections + i);
+      fdstate_open(newfd, g_connections + i, 0);
 
       /* Prepare connection info block */
       reset_info_block(g_connections + i);
@@ -310,7 +311,7 @@ static void handle_accept(int64_t serversocket) {
       g_connections[i].port = port;
       g_connections[i].fd   = newfd;
 
-      io_setcookie(newfd, g_connections + i);
+      fdstate_open(newfd, g_connections + i, 0);
 
       /* We expect the connecting side to begin with its tracker_id */
       io_wantread(newfd);
@@ -325,7 +326,7 @@ static void handle_read(int64_t peersocket) {
   int         i;
   int64_t     datalen;
   uint32_t    tracker_id;
-  proxy_peer *peer = io_getcookie(peersocket);
+  proxy_peer *peer = fdstate_cookie(peersocket);
 
   if (!peer) {
     /* Can't happen ;) */
@@ -391,7 +392,7 @@ static void handle_read(int64_t peersocket) {
 
 /* Can write new sync data to the stream */
 static void handle_write(int64_t peersocket) {
-  proxy_peer *peer = io_getcookie(peersocket);
+  proxy_peer *peer = fdstate_cookie(peersocket);
 
   if (!peer) {
     /* Can't happen ;) */
@@ -464,7 +465,7 @@ static void server_mainloop() {
 
     /* Loop over readable sockets */
     while ((sock = io_canread()) != -1) {
-      const void *cookie = io_getcookie(sock);
+      const void *cookie = fdstate_cookie(sock);
       if ((uintptr_t)cookie == FLAG_SERVERSOCKET)
         handle_accept(sock);
       else
@@ -493,7 +494,7 @@ static int64_t ot_try_bind(ot_ip6 ip, uint16_t port) {
   if (!io_fd(sock))
     panic("io_fd");
 
-  io_setcookie(sock, (void *)FLAG_SERVERSOCKET);
+  fdstate_open(sock, (void *)FLAG_SERVERSOCKET, 0);
   io_wantread(sock);
   return sock;
 }
@@ -535,6 +536,8 @@ int main(int argc, char **argv) {
   int              scanon = 1, lbound = 0, sbound = 0;
 
   random_init();
+  if (fdstate_init() == -1)
+    panic("fdstate_init");
   g_tracker_id = random_uint32();
 
   while (scanon) {