Generalize the worker setup/runloop/teardown logic by adding a smol ot_worker helper

1 files changed, 5 insertions, 61 deletions

diff --git a/ot_tcp.c b/ot_tcp.c
index 47697ca..4a5008b 100644
--- a/ot_tcp.c
+++ b/ot_tcp.c
@@ -20,6 +20,7 @@
 #include "ot_http.h"
 #include "ot_mutex.h"
 #include "ot_socket.h"
+#include "ot_worker.h"
 #include "ot_stats.h"
 #include "ot_tcp.h"
 #include "trackerlogic.h"
@@ -164,75 +165,18 @@ void tcp_accept(const int64_t serversocket, struct ot_workstruct *ws) {
   }
 }
 
-/* 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;
-  }
-
+static void tcp_run(struct ot_worker *me) {
   while (g_opentracker_running)
-    tcp_accept(me->sock, &ws);
-
-  workstruct_deinit(&ws);
-  atomic_store(&me->left, 1);
-  return NULL;
+    tcp_accept(me->sock, &me->ws);
 }
 
 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;
+  if (worker_start(tcp_run, sock) == -1)
+    fprintf(stderr, "Warning: could not install a tcp worker, this socket stays with the workers already running.\n");
 }