1 files changed, 238 insertions, 0 deletions

diff --git a/ot_tcp.c b/ot_tcp.c
new file mode 100644
index 0000000..47697ca
--- /dev/null
+++ b/ot_tcp.c
@@ -0,0 +1,238 @@
+/* This software was written by Dirk Engling <erdgeist@erdgeist.org>
+   It is considered beerware. Prost. Skol. Cheers or whatever.
+
+   $id$ */
+
+/* System */
+#include <errno.h>
+#include <pthread.h>
+#include <stdatomic.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <unistd.h>
+
+/* Opentracker */
+#include "ot_batch.h"
+#include "ot_buf.h"
+#include "ot_event.h"
+#include "ot_fdstate.h"
+#include "ot_http.h"
+#include "ot_mutex.h"
+#include "ot_socket.h"
+#include "ot_stats.h"
+#include "ot_tcp.h"
+#include "trackerlogic.h"
+
+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;
+}
+
+void tcp_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;
+  }
+}
+
+void tcp_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_set(sock, cookie, 0);
+  else if (keep_alive) {
+    /* The reply is out and the client asked to keep the connection, so go
+       back to waiting for the next request */
+    http_interest_set(sock, cookie, EVENT_IN);
+    fdstate_set_expiry(sock, time_future_seconds(OT_CLIENT_TIMEOUT));
+  } else
+    http_close(sock);
+}
+
+/* The fast path, assuming that the complete request has usually arrived by the time a
+   connection is accepted, so it is answered here outside the event loop. The more load
+   a tracker has, the more likely it is that by the time it accepts, the roundtrip to
+   the peer has completed.
+   If the request could not be completely handled, the stack cookie is promoted to heap
+   and the connection handed over to the event loop to finish */
+void tcp_accept(const int64_t serversocket, struct ot_workstruct *ws) {
+  int64_t  sock;
+  ot_ip6   ip;
+  uint16_t port;
+
+  while ((sock = socket_accept(serversocket, ip, &port)) != -1) {
+    struct http_data cookie;
+    uint32_t         generation;
+
+    memset(&cookie, 0, sizeof(cookie));
+    memcpy(cookie.ip, ip, sizeof(ot_ip6));
+    cookie.interest = EVENT_IN;
+    cookie.flag     = STRUCT_HTTP_FLAG_ONSTACK;
+
+    generation = fdstate_open(sock, &cookie, time_future_seconds(OT_CLIENT_TIMEOUT));
+    if (!fdstate_claim((int)sock, generation)) {
+      fdstate_close(sock);
+      event_close((int)sock);
+      continue;
+    }
+
+    stats_issue_event(ws, EVENT_ACCEPT, FLAG_TCP, (uintptr_t)ip);
+
+    tcp_read(sock, ws);
+
+    /* Gone already if it was answered and closed inline */
+    if (fdstate_generation((int)sock) != generation)
+      continue;
+
+    /* 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;
+}