Make tcp worker multithreaded, too

This shamelessly steals the model established in the UDP workers, i.e.
if the kernel is able to bind multiple listening sockets to the same
endpoint, we will share them across threads. If it can't, all threads
will accept on the same socket.

The fast path handles most of the connections, only the rare case of
no data yet on the socket is offloaded to the slow path (which used to
be the mainloop) for completion. This scales nicely, as the higher the
load is, the higher chances are that an accept happens only after the
network roundtrip to the peer completes, leading to more connections
that already have data waiting for us.

We also consolidate the workstruct init and deinit code, preparation
for later unifying all thread state structs for all protocols (mca,
udc and tcp).

The config file gains a per tcp connection worker count and some docs
explaining what this knob buys and costs, nudging the user to either
keep it 0 or dial it up to at least 2.

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
 */