1 files changed, 50 insertions, 152 deletions

diff --git a/opentracker.c b/opentracker.c
index 6160696..2f2746d 100644
--- a/opentracker.c
+++ b/opentracker.c
@@ -29,12 +29,14 @@
 #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"
 #include "ot_fdstate.h"
 #include "ot_socket.h"
 #include "ot_stats.h"
+#include "ot_tcp.h"
 #include "ot_udp.h"
 #include "trackerlogic.h"
 
@@ -96,129 +98,6 @@ static void install_signal_handlers(void) {
 }
 
 
-static ssize_t header_complete(char *request, ssize_t byte_count) {
-  ssize_t i = 0, state = 0;
-
-  for (i = 1; i < byte_count; i += 2)
-    if (request[i] <= 13) {
-      i--;
-      for (state = 0; i < byte_count; ++i) {
-        char c = request[i];
-        if (c == '\r' || c == '\n')
-          state = (state >> 2) | ((c << 6) & 0xc0);
-        else
-          break;
-        if (state >= 0xa0 || state == 0x99)
-          return i + 1;
-      }
-    }
-  return 0;
-}
-
-static void handle_read(const int64_t sock, struct ot_workstruct *ws) {
-  struct http_data *cookie     = fdstate_cookie(sock);
-  ssize_t           byte_count = read((int)sock, ws->inbuf, G_INBUF_SIZE);
-
-  if (byte_count == 0 || (byte_count == -1 && errno != EAGAIN && errno != EWOULDBLOCK && errno != EINTR)) {
-    http_close(sock);
-    return;
-  }
-
-  if (byte_count == -1)
-    return;
-
-  ws->request      = NULL;
-  ws->request_size = 0;
-  ws->header_size  = 0;
-  ws->keep_alive   = 0;
-
-  /* If we get the whole request in one packet, handle it without copying */
-  if (!cookie->request.data) {
-    if ((ws->header_size = header_complete(ws->inbuf, byte_count))) {
-      ws->request      = ws->inbuf;
-      ws->request_size = byte_count;
-      http_handle_request(sock, ws);
-    } else if (buf_cat(&cookie->request, ws->inbuf, (size_t)byte_count))
-      http_close(sock);
-    return;
-  }
-
-  if (buf_cat(&cookie->request, ws->inbuf, (size_t)byte_count) || cookie->request.size > G_INBUF_SIZE) {
-    http_issue_error(sock, ws, CODE_HTTPERROR_500);
-    return;
-  }
-
-  while ((ws->header_size = header_complete(cookie->request.data, (ssize_t)cookie->request.size))) {
-    ws->request      = cookie->request.data;
-    ws->request_size = (ssize_t)cookie->request.size;
-    http_handle_request(sock, ws);
-#ifdef WANT_KEEPALIVE
-    if (!ws->keep_alive)
-#endif
-      return;
-  }
-}
-
-static void handle_write(const int64_t sock) {
-  struct http_data *cookie     = fdstate_cookie(sock);
-  int               chunked    = 0;
-  int               keep_alive = 0;
-
-  if (cookie) {
-    chunked    = (cookie->flag & STRUCT_HTTP_FLAG_CHUNKED_IN_TRANSFER) != 0;
-    keep_alive = (cookie->flag & STRUCT_HTTP_FLAG_KEEPALIVE) != 0;
-
-    if (cookie->batch) {
-      if (batch_send(sock, cookie->batch) == BATCH_FAILED) {
-        http_close(sock);
-        return;
-      }
-      if (batch_pending(cookie->batch))
-        return;
-    }
-  }
-
-  /* In a chunked transfer, wait for the next piece once everything queued has gone out */
-  if (chunked)
-    http_interest_drop(sock, cookie, EVENT_OUT);
-  else if (keep_alive) {
-    /* The reply is out and the client asked to keep the connection, so stop
-       waiting to write and go back to waiting for the next request */
-    http_interest_drop(sock, cookie, EVENT_OUT);
-    http_interest_add(sock, cookie, EVENT_IN);
-    fdstate_set_expiry(sock, time_future_seconds(OT_CLIENT_TIMEOUT));
-  } else
-    http_close(sock);
-}
-
-static void handle_accept(const int64_t serversocket, struct ot_workstruct *ws) {
-  struct http_data *cookie;
-  int64_t           sock;
-  ot_ip6            ip;
-  uint16_t          port;
-
-  while ((sock = socket_accept(serversocket, ip, &port)) != -1) {
-
-    socket_set_blocking((int)sock, 0);
-
-    if (!(cookie = (struct http_data *)calloc(1, sizeof(struct http_data)))) {
-      close((int)sock);
-      continue;
-    }
-    memcpy(cookie->ip, ip, sizeof(ot_ip6));
-    cookie->interest = EVENT_IN;
-
-    if (event_add((int)sock, EVENT_IN, fdstate_open(sock, cookie, time_future_seconds(OT_CLIENT_TIMEOUT))) == -1) {
-      fdstate_close(sock);
-      free(cookie);
-      close((int)sock);
-      continue;
-    }
-
-    stats_issue_event(ws, EVENT_ACCEPT, FLAG_TCP, (uintptr_t)ip);
-  }
-}
-
 /* How many ready descriptors one wait hands back at a time */
 #define OT_EVENT_BATCH 512
 
@@ -231,20 +110,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;
@@ -264,22 +132,24 @@ static void *server_mainloop(void *args) {
       cookie = fdstate_cookie((int)sock);
 
       if ((intptr_t)cookie == FLAG_TCP)
-        handle_accept(sock, &ws);
+        tcp_accept(sock, &ws);
       else if ((intptr_t)cookie == FLAG_UDP)
         handle_udp6(sock, &ws);
       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)
-          handle_read(sock, &ws);
+          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)
-          handle_write(sock);
+          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) {
@@ -295,13 +165,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;
 }
 
@@ -310,7 +174,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;
 }
@@ -342,6 +206,7 @@ static void listen_bind_one(ot_listen *entry) {
   if (entry->proto == FLAG_TCP) {
     if ((entry->sock = socket_tcp_listen(entry->ip, entry->port, sharing)) == -1)
       panic("socket_tcp_listen");
+    socket_defer_accept((int)entry->sock);
   } else {
     if ((entry->sock = socket_udp_bind(entry->ip, entry->port, sharing)) == -1)
       panic("socket_udp_bind");
@@ -365,10 +230,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);
@@ -529,9 +425,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