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, 40 insertions, 24 deletions

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