12 files changed, 294 insertions, 59 deletions

diff --git a/man4/opentracker.conf.4 b/man4/opentracker.conf.4
index aa82e1d..1bd1c7a 100644
--- a/man4/opentracker.conf.4
+++ b/man4/opentracker.conf.4
@@ -42,6 +42,14 @@ in the thread that also serves TCP. See
 .Sx TUNING
 before raising it.
 .Pp
+.It listen.tcp.workers Ar threads
+Specifies how many threads accept and answer HTTP connections, for every
+listen statement that does not carry a
+.Cm workers
+option of its own. Defaults to 0, which serves them from the event loop. See
+.Sx TUNING
+before raising it.
+.Pp
 .It listen.shared Ar boolean
 Permits another process to bind the same address, for every listen statement
 that does not carry a
@@ -101,6 +109,14 @@ group only where the answer is yes.
 in the source distribution sweeps the setting against a running tracker.
 Measure on the machine that will carry the traffic.
 .Pp
+A TCP worker accepts a connection and answers it on its own stack, without
+the event loop seeing it at all; only a connection it cannot finish is handed
+over. One worker serves less than none, because accepting leaves the event
+loop's thread without anything gained in return, so the setting is worth
+either 0 or at least 2.
+.Pa tests/http_load.c
+in the source distribution drives the tracker over HTTP to measure it.
+.Pp
 On FreeBSD the
 .Xr accf_data 9
 filter has the kernel hold a new connection until data arrives on it, so the
diff --git a/opentracker.c b/opentracker.c
index 2eaf988..e5efff5 100644
--- a/opentracker.c
+++ b/opentracker.c
@@ -109,20 +109,9 @@ static void *server_mainloop(void *args) {
 
   (void)args;
 
-  /* Initialize our "thread local storage" */
-  memset(&ws, 0, sizeof(ws));
-  stats_counters_register(&ws);
-  ws.inbuf  = malloc(G_INBUF_SIZE);
-  ws.outbuf = malloc(G_OUTBUF_SIZE);
-#ifdef _DEBUG_HTTPERROR
-  ws.debugbuf = malloc(G_DEBUGBUF_SIZE);
-#endif
-
-  if (!ws.inbuf || !ws.outbuf)
+  if (workstruct_init(&ws) == -1)
     panic("Initializing worker failed");
 
-  random_bytes(ws.rand48_state, sizeof(ws.rand48_state));
-
   while (g_opentracker_running) {
     ot_event ready[OT_EVENT_BATCH];
     int64_t  sock;
@@ -173,13 +162,7 @@ static void *server_mainloop(void *args) {
 
   /* This work struct is about to go out of scope, so hand its tally over
      before anything can read it there */
-  stats_counters_retire(&ws);
-
-  free(ws.inbuf);
-  free(ws.outbuf);
-#ifdef _DEBUG_HTTPERROR
-  free(ws.debugbuf);
-#endif
+  workstruct_deinit(&ws);
   return 0;
 }
 
@@ -188,7 +171,7 @@ static void *server_mainloop(void *args) {
    member asked for it before binding, the first one included, so whether
    this entry will have workers has to be known here. */
 static socket_sharing listen_sharing_for(const ot_listen *entry) {
-  if (entry->proto == FLAG_UDP && entry->workers > 1 && socket_can_balance())
+  if (entry->proto != FLAG_MCA && entry->workers > 1 && socket_can_balance())
     return SOCKET_BALANCED;
   return entry->shared ? SOCKET_SHARED : SOCKET_EXCLUSIVE;
 }
@@ -244,10 +227,41 @@ static void listen_bind_all(void) {
     listen_bind_one(entries + i);
 }
 
-/* The socket the event loop was using becomes the first worker's. Where the
-   kernel spreads datagrams, every further worker binds one of its own, so
-   none of them share a receive queue.
-   All others read from the first */
+/* The socket the event loop was using is passed to the first worker.
+   If start up tests show that the kernel can spread connections over
+   a group bound to one address, every worker binds one of its own.
+   If not, all workers call accept on the first. */
+static void listen_start_tcp_workers(void) {
+  size_t     count, i;
+  ot_listen *entries = config_listen_entries(&count);
+
+  for (i = 0; i < count; ++i) {
+    ot_listen *entry   = entries + i;
+    int        balance = listen_sharing_for(entry) == SOCKET_BALANCED;
+    int        worker;
+
+    if (entry->proto != FLAG_TCP || entry->workers <= 0)
+      continue;
+
+    event_del((int)entry->sock);
+    tcp_worker_install(entry->sock);
+
+    for (worker = 1; worker < entry->workers; ++worker) {
+      int64_t sock = entry->sock;
+
+      if (balance) {
+        sock = socket_tcp_listen(entry->ip, entry->port, SOCKET_BALANCED);
+        if (sock == -1) {
+          fprintf(stderr, "Warning: could not bind another socket for tcp worker %d: %s. Sharing the first instead.\n", worker,
+                  strerror(errno));
+          sock = entry->sock;
+        }
+      }
+      tcp_worker_install(sock);
+    }
+  }
+}
+
 static void listen_start_udp_workers(void) {
   size_t     count, i;
   ot_listen *entries = config_listen_entries(&count);
@@ -408,9 +422,11 @@ int main(int argc, char **argv) {
 
   install_signal_handlers();
 
+  listen_start_tcp_workers();
   listen_start_udp_workers();
 
   server_mainloop(0);
+  tcp_workers_deinit();
   udp_workers_deinit();
 
   /* No thread is serving any more, so the teardown can run where a debugger
diff --git a/opentracker.conf.sample b/opentracker.conf.sample
index 1d535e8..f075703 100644
--- a/opentracker.conf.sample
+++ b/opentracker.conf.sample
@@ -21,9 +21,9 @@
 #
 #      A listen statement takes options after the address:
 #
-#        workers=<count>  handle this udp socket in blocking reads in that
-#                         many dedicated worker threads instead of in the
-#                         event loop. Defaults to 1. Ignored for tcp.
+#        workers=<count>  handle this socket in that many dedicated worker
+#                         threads instead of in the event loop. Defaults to 1
+#                         for udp and 0 for tcp.
 #        shared           allow another process to bind the same address and
 #                         let the kernel share the traffic. Only useful if you
 #                         deliberately run a second tracker there.
@@ -35,6 +35,7 @@
 #      anywhere.
 #
 # listen.udp.workers 4
+# listen.tcp.workers 4
 # listen.shared 1
 #
 #      Raising the worker count pays off where the kernel spreads arriving
diff --git a/ot_fdstate.c b/ot_fdstate.c
index 7d742da..09daf50 100644
--- a/ot_fdstate.c
+++ b/ot_fdstate.c
@@ -91,7 +91,7 @@ uint32_t fdstate_open(int fd, void *cookie, ot_time expiry) {
   }
 
   word = atomic_load_explicit(&slot->state, memory_order_relaxed);
-  atomic_store_explicit(&slot->cookie, cookie, memory_order_relaxed);
+  atomic_store_explicit(&slot->cookie, cookie, memory_order_release);
   atomic_store_explicit(&slot->expiry, expiry, memory_order_relaxed);
   /* Publishes the two above to whoever claims the slot next */
   atomic_store_explicit(&slot->state, FD_WORD(FD_GEN(word), FD_IDLE), memory_order_release);
@@ -183,14 +183,17 @@ int fdstate_next_expired(ot_time now, int *cursor) {
 void *fdstate_cookie(int fd) {
   ot_fdslot *slot = fdstate_slot(fd);
 
-  return slot ? atomic_load_explicit(&slot->cookie, memory_order_relaxed) : NULL;
+  /* Pairs with the release in fdstate_set_cookie and fdstate_open */
+  return slot ? atomic_load_explicit(&slot->cookie, memory_order_acquire) : NULL;
 }
 
 void fdstate_set_cookie(int fd, void *cookie) {
   ot_fdslot *slot = fdstate_slot(fd);
 
+  /* Publishes what the cookie points at to whoever reads it next, which may
+     be another thread entirely */
   if (slot)
-    atomic_store_explicit(&slot->cookie, cookie, memory_order_relaxed);
+    atomic_store_explicit(&slot->cookie, cookie, memory_order_release);
 }
 
 void fdstate_set_expiry(int fd, ot_time expiry) {
diff --git a/ot_http.c b/ot_http.c
index bb6ff24..602ff94 100644
--- a/ot_http.c
+++ b/ot_http.c
@@ -39,22 +39,30 @@ ssize_t      g_stats_path_len;
 
 enum { SUCCESS_HTTP_HEADER_LENGTH = 80, SUCCESS_HTTP_SIZE_OFF = 17 };
 
-int http_adopt(const int64_t sock, struct http_data *cookie) {
-  struct http_data *heap = malloc(sizeof(*heap));
+struct http_data *http_adopt(const int64_t sock) {
+  struct http_data *cookie = fdstate_cookie(sock), *heap;
+  int               interest;
 
-  if (!heap)
-    return -1;
+  if (!cookie || !(cookie->flag & STRUCT_HTTP_FLAG_ONSTACK))
+    return cookie;
+
+  if (!(heap = malloc(sizeof(*heap))))
+    return NULL;
 
   *heap = *cookie;
   heap->flag &= ~STRUCT_HTTP_FLAG_ONSTACK;
+
+  /* Publishing the cookie is what lets another thread reach this connection,
+     so nothing may touch it afterwards, not even to read what it asked for */
+  interest = heap->interest;
   fdstate_set_cookie((int)sock, heap);
 
-  if (event_add((int)sock, heap->interest, fdstate_generation((int)sock)) == -1) {
+  if (event_add((int)sock, interest, fdstate_generation((int)sock)) == -1) {
     fdstate_set_cookie((int)sock, cookie);
     free(heap);
-    return -1;
+    return NULL;
   }
-  return 0;
+  return heap;
 }
 
 /* Give up a connection: the one place that knows everything hanging off it. */
@@ -348,13 +356,17 @@ static ssize_t http_handle_stats(const int64_t sock, struct ot_workstruct *ws, c
     }
 #endif
 #endif
-    /* 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 */
     fdstate_set_expiry(sock, 0);
-    fullscrape_deliver(sock, format);
     http_interest_set(sock, cookie, 0);
+
+    /* Everything this connection needs has been written, so it can be given
+       a home of its own and only then be made reachable by a worker thread */
+    if (!http_adopt(sock))
+      HTTPERROR_500;
+    fullscrape_deliver(sock, format);
     return ws->reply_size = -2;
   }
 #endif
@@ -366,6 +378,9 @@ static ssize_t http_handle_stats(const int64_t sock, struct ot_workstruct *ws, c
 
     /* Complex stats also include expensive memory debugging tools */
     fdstate_set_expiry(sock, 0);
+
+    if (!http_adopt(sock))
+      HTTPERROR_500;
     stats_deliver(sock, mode);
     return ws->reply_size = -2;
   }
@@ -448,8 +463,11 @@ static ssize_t http_handle_fullscrape(const int64_t sock, struct ot_workstruct *
   cookie->flag |= STRUCT_HTTP_FLAG_WAITINGFORTASK | STRUCT_HTTP_FLAG_CHUNKED;
   /* Clients waiting for us should not easily timeout */
   fdstate_set_expiry(sock, 0);
-  fullscrape_deliver(sock, format);
   http_interest_set(sock, cookie, 0);
+
+  if (!http_adopt(sock))
+    HTTPERROR_500;
+  fullscrape_deliver(sock, format);
   return ws->reply_size = -2;
 }
 #endif
diff --git a/ot_http.h b/ot_http.h
index 82d645b..9cacd21 100644
--- a/ot_http.h
+++ b/ot_http.h
@@ -40,9 +40,10 @@ void           http_interest_set(const int64_t s, struct http_data *cookie, int
 /* Close a connection and release everything that belongs to it */
 void           http_close(const int64_t s);
 
-/* Promote a stack cookie to a heap one for persistence and handing it over
-   to the event loop. */
-int            http_adopt(const int64_t s, struct http_data *cookie);
+/* Promote a stack cookie to a heap one, for persistence and for handing it
+   over to the event loop or to another thread. Does nothing to one that is
+   already on the heap. Returns the connection's state, or NULL. */
+struct http_data *http_adopt(const int64_t s);
 
 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);
diff --git a/ot_tcp.c b/ot_tcp.c
index 8647a47..47697ca 100644
--- a/ot_tcp.c
+++ b/ot_tcp.c
@@ -5,6 +5,10 @@
 
 /* System */
 #include <errno.h>
+#include <pthread.h>
+#include <stdatomic.h>
+#include <stdio.h>
+#include <stdlib.h>
 #include <string.h>
 #include <unistd.h>
 
@@ -149,9 +153,86 @@ void tcp_accept(const int64_t serversocket, struct ot_workstruct *ws) {
     if (fdstate_generation((int)sock) != generation)
       continue;
 
-    if (http_adopt(sock, &cookie) == -1)
+    /* Handling it may already have given it a heap home and handed it to
+       another thread, and then it is no longer this frame's to touch */
+    if (fdstate_cookie((int)sock) != &cookie)
+      fdstate_release((int)sock);
+    else if (!http_adopt(sock))
       http_close(sock);
     else
       fdstate_release((int)sock);
   }
 }
+
+/* One of these per worker, each allocated on its own so that growing the
+   array below never moves the one a running worker holds. */
+struct tcp_worker {
+  pthread_t   thread;
+  int64_t     sock;
+  _Atomic int left;
+};
+
+static struct tcp_worker **g_workers;
+static size_t              g_worker_count;
+
+/* A worker blocks in accept and nowhere else: the connection it is handed is
+   non blocking, so a client that stops talking is given to the event loop
+   rather than keeping this thread. */
+static void *tcp_worker(void *args) {
+  struct tcp_worker   *me = args;
+  struct ot_workstruct ws;
+
+  if (workstruct_init(&ws) == -1) {
+    fprintf(stderr, "Warning: a tcp worker could not get its buffers and will not run.\n");
+    atomic_store(&me->left, 1);
+    return NULL;
+  }
+
+  while (g_opentracker_running)
+    tcp_accept(me->sock, &ws);
+
+  workstruct_deinit(&ws);
+  atomic_store(&me->left, 1);
+  return NULL;
+}
+
+void tcp_worker_install(int64_t sock) {
+  struct tcp_worker **grown = realloc(g_workers, (g_worker_count + 1) * sizeof(*g_workers));
+  struct tcp_worker  *worker;
+
+  if (!grown || !(worker = calloc(1, sizeof(*worker)))) {
+    fprintf(stderr, "Warning: out of memory installing a tcp worker, this socket stays with the workers already running.\n");
+    g_workers = grown ? grown : g_workers;
+    return;
+  }
+  g_workers    = grown;
+  worker->sock = sock;
+
+  socket_set_blocking((int)sock, 1);
+  socket_defer_accept((int)sock);
+
+  if (pthread_create(&worker->thread, NULL, tcp_worker, worker))
+    free(worker);
+  else
+    g_workers[g_worker_count++] = worker;
+}
+
+void tcp_workers_deinit(void) {
+  size_t i;
+
+  for (i = 0; i < g_worker_count; ++i) {
+    struct tcp_worker *worker = g_workers[i];
+
+    /* Repeated because a signal only interrupts a call already under way */
+    while (!atomic_load(&worker->left)) {
+      pthread_kill(worker->thread, OT_SIGNAL_WAKE);
+      usleep(1000);
+    }
+    pthread_join(worker->thread, NULL);
+    free(worker);
+  }
+
+  free(g_workers);
+  g_workers      = NULL;
+  g_worker_count = 0;
+}
diff --git a/ot_tcp.h b/ot_tcp.h
index b0a93f4..96b6ec1 100644
--- a/ot_tcp.h
+++ b/ot_tcp.h
@@ -19,4 +19,10 @@ void tcp_accept(const int64_t serversocket, struct ot_workstruct *ws);
 void tcp_read(const int64_t sock, struct ot_workstruct *ws);
 void tcp_write(const int64_t sock);
 
+/* One thread accepting on one socket. Called once per worker. */
+void tcp_worker_install(int64_t sock);
+
+/* Stop every worker installed above and wait for it to be gone */
+void tcp_workers_deinit(void);
+
 #endif
diff --git a/ot_udp.c b/ot_udp.c
index 1ec5148..00ec566 100644
--- a/ot_udp.c
+++ b/ot_udp.c
@@ -220,26 +220,17 @@ int handle_udp6(int64_t serversocket, struct ot_workstruct *ws) {
 static void *udp_worker(void *args) {
   struct udp_worker   *me = args;
   struct ot_workstruct ws;
-  memset(&ws, 0, sizeof(ws));
 
-  stats_counters_register(&ws);
-  ws.inbuf  = malloc(G_INBUF_SIZE);
-  ws.outbuf = malloc(G_OUTBUF_SIZE);
-#ifdef _DEBUG_HTTPERROR
-  ws.debugbuf = malloc(G_DEBUGBUF_SIZE);
-#endif
+  if (workstruct_init(&ws) == -1) {
+    fprintf(stderr, "Warning: a udp worker could not get its buffers and will not run.\n");
+    atomic_store(&me->left, 1);
+    return NULL;
+  }
 
   while (g_opentracker_running)
     handle_udp6(me->sock, &ws);
 
-  stats_counters_retire(&ws);
-
-  free(ws.inbuf);
-  free(ws.outbuf);
-#ifdef _DEBUG_HTTPERROR
-  free(ws.debugbuf);
-#endif
-
+  workstruct_deinit(&ws);
   atomic_store(&me->left, 1);
   return NULL;
 }
diff --git a/tests/integration/test_tcpworkers.py b/tests/integration/test_tcpworkers.py
new file mode 100644
index 0000000..e43612a
--- /dev/null
+++ b/tests/integration/test_tcpworkers.py
@@ -0,0 +1,67 @@
+"""The tracker with its tcp side handed to worker threads.
+
+With workers configured, a connection is accepted and answered on a worker's
+own stack. Anything it cannot finish there is copied to the heap and handed
+to the event loop, which is the one place where a worker and the main thread
+touch the same structures, so these drive all three outcomes: answered
+inline, held open, and parked waiting for another thread entirely.
+"""
+
+import threading
+import unittest
+import urllib.parse
+
+from harness import TrackerTestCase, bdecode
+
+WORKERS = 4
+
+
+class TcpWorkers(TrackerTestCase):
+    config = "listen.tcp.workers %d\n" % WORKERS
+
+    def announce_path(self, name, port=6881):
+        return "/announce?info_hash=%s&peer_id=%s&port=%d&left=0" % (
+            urllib.parse.quote_from_bytes(self.hash_for(name), safe=""),
+            "x" * 20,
+            port,
+        )
+
+    def test_many_at_once_are_all_answered(self):
+        """Every worker gets something to do, and the swarm adds up."""
+        info_hash = self.hash_for("workers-swarm")
+        errors = []
+
+        def announce(index):
+            try:
+                self.tracker.announce(info_hash, peer_id=b"-OT0001-%012d" % index, port=7000 + index, left=0)
+            except Exception as problem:  # noqa: BLE001 - reported, not swallowed
+                errors.append("%d: %r" % (index, problem))
+
+        threads = [threading.Thread(target=announce, args=(i,)) for i in range(32)]
+        for thread in threads:
+            thread.start()
+        for thread in threads:
+            thread.join()
+
+        self.assertEqual(errors, [], "\n".join(errors))
+        self.assertEqual(self.tracker.scrape(info_hash)[b"files"][info_hash][b"complete"], 32)
+
+    def test_a_kept_connection_outlives_the_worker_that_took_it(self):
+        """Keep-alive is the plain case of a connection a worker hands on."""
+        connection = self.tracker.connect()
+        try:
+            for i in range(4):
+                connection.send(self.announce_path("workers-kept"))
+                status, _, body = connection.reply()
+                self.assertIn(b"200", status)
+                self.assertIn(b"complete", bdecode(body))
+        finally:
+            connection.close()
+
+    def test_a_full_scrape_asked_for_by_a_worker_still_arrives(self):
+        """The reply is built by another thread, so the worker parks the
+           connection waiting for nothing at all before handing it over."""
+        for i in range(20):
+            self.tracker.announce(bytes([i]) + b"\0" * 19, port=7100 + i, left=0)
+        body = self.tracker.get("/scrape")
+        self.assertEqual(len(bdecode(body)[b"files"]), 20)
diff --git a/trackerlogic.c b/trackerlogic.c
index 477c4a6..fc0fa56 100644
--- a/trackerlogic.c
+++ b/trackerlogic.c
@@ -567,6 +567,35 @@ void trackerlogic_init(void) {
   stats_init();
 }
 
+int workstruct_init(struct ot_workstruct *ws) {
+  memset(ws, 0, sizeof(*ws));
+  ws->inbuf  = malloc(G_INBUF_SIZE);
+  ws->outbuf = malloc(G_OUTBUF_SIZE);
+#ifdef _DEBUG_HTTPERROR
+  ws->debugbuf = malloc(G_DEBUGBUF_SIZE);
+#endif
+  if (!ws->inbuf || !ws->outbuf)
+    return -1;
+
+  /* Picking which peers to hand out starts here, so two workers seeded alike
+     would hand out alike */
+  random_bytes(ws->rand48_state, sizeof(ws->rand48_state));
+
+  stats_counters_register(ws);
+  return 0;
+}
+
+void workstruct_deinit(struct ot_workstruct *ws) {
+  /* The tally goes to the global before this struct goes out of scope */
+  stats_counters_retire(ws);
+
+  free(ws->inbuf);
+  free(ws->outbuf);
+#ifdef _DEBUG_HTTPERROR
+  free(ws->debugbuf);
+#endif
+}
+
 void trackerlogic_deinit(void) {
   int    delta_torrentcount = 0;
   size_t bucket, j;
diff --git a/trackerlogic.h b/trackerlogic.h
index d269267..53edabd 100644
--- a/trackerlogic.h
+++ b/trackerlogic.h
@@ -236,6 +236,12 @@ struct ot_workstruct {
   int      keep_alive;
 };
 
+/* Everything a thread that serves requests needs for itself. Set up and torn
+   down in one place so that a new kind of worker cannot miss a piece.
+   Returns -1 if the buffers could not be had. */
+int  workstruct_init(struct ot_workstruct *ws);
+void workstruct_deinit(struct ot_workstruct *ws);
+
 /*
    Exported functions
 */