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