1 files changed, 46 insertions, 27 deletions

diff --git a/opentracker.c b/opentracker.c
index 2eaf988..644e99c 100644
--- a/opentracker.c
+++ b/opentracker.c
@@ -29,6 +29,7 @@
 #include "ot_ip.h"
 #include "ot_livesync.h"
 #include "ot_mutex.h"
+#include "ot_workqueue.h"
 #include "ot_random.h"
 #include "ot_scan.h"
 #include "ot_event.h"
@@ -37,6 +38,7 @@
 #include "ot_stats.h"
 #include "ot_tcp.h"
 #include "ot_udp.h"
+#include "ot_worker.h"
 #include "trackerlogic.h"
 
 /* Globals */
@@ -109,20 +111,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;
@@ -148,16 +139,18 @@ static void *server_mainloop(void *args) {
       else if ((intptr_t)cookie == FLAG_SELFPIPE)
         while (read((int)sock, ws.inbuf, G_INBUF_SIZE) > 0)
           ;
-      else {
+      else if (fdstate_claim((int)sock, ready[i].generation)) {
         if (ready[i].what & EVENT_IN)
           tcp_read(sock, &ws);
         /* The read may have closed it, so look again before writing */
         if ((ready[i].what & EVENT_OUT) && fdstate_generation((int)sock) == ready[i].generation)
           tcp_write(sock);
+        if (fdstate_generation((int)sock) == ready[i].generation)
+          fdstate_release((int)sock);
       }
     }
 
-    while ((sock = mutex_workqueue_popresult(&iovec_entries, &iovector, &is_partial)) != -1)
+    while ((sock = workqueue_popresult(&iovec_entries, &iovector, &is_partial)) != -1)
       http_sendiovecdata(sock, &ws, iovec_entries, iovector, is_partial);
 
     if (time_now_seconds() > next_timeout_check) {
@@ -173,13 +166,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 +175,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 +231,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,10 +426,11 @@ int main(int argc, char **argv) {
 
   install_signal_handlers();
 
+  listen_start_tcp_workers();
   listen_start_udp_workers();
 
   server_mainloop(0);
-  udp_workers_deinit();
+  worker_stop_all();
 
   /* No thread is serving any more, so the teardown can run where a debugger
      and the sanitizers can follow it */