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.

1 files changed, 82 insertions, 1 deletions

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;
+}