Refactor all tcp specific code into the ot_tcp module to prepare for multiple concurrent workers

4 files changed, 185 insertions, 141 deletions

diff --git a/Makefile b/Makefile
index 2e52367..8f4dcf6 100644
--- a/Makefile
+++ b/Makefile
@@ -69,8 +69,8 @@ CFLAGS_libowfat=-I$(LIBOWFAT_HEADERS)
 LDFLAGS_libowfat=-L$(LIBOWFAT_LIBRARY) -lowfat
 
 BINARY =opentracker
-HEADERS=trackerlogic.h ot_urlencoded.h ot_batch.h ot_buf.h ot_event.h ot_fdstate.h ot_fmt.h ot_hex.h ot_ip.h ot_config.h ot_scan.h ot_socket.h ot_mutex.h ot_random.h ot_stats.h ot_vector.h ot_clean.h ot_udp.h ot_iovec.h ot_fullscrape.h ot_accesslist.h ot_http.h ot_livesync.h ot_rijndael.h
-SOURCES=opentracker.c trackerlogic.c ot_urlencoded.c ot_batch.c ot_buf.c ot_event.c ot_fdstate.c ot_fmt.c ot_hex.c ot_ip.c ot_config.c ot_scan.c ot_socket.c ot_mutex.c ot_random.c ot_stats.c ot_vector.c ot_clean.c ot_udp.c ot_iovec.c ot_fullscrape.c ot_accesslist.c ot_http.c ot_livesync.c ot_rijndael.c
+HEADERS=trackerlogic.h ot_urlencoded.h ot_batch.h ot_buf.h ot_event.h ot_fdstate.h ot_fmt.h ot_hex.h ot_ip.h ot_config.h ot_scan.h ot_socket.h ot_tcp.h ot_mutex.h ot_random.h ot_stats.h ot_vector.h ot_clean.h ot_udp.h ot_iovec.h ot_fullscrape.h ot_accesslist.h ot_http.h ot_livesync.h ot_rijndael.h
+SOURCES=opentracker.c trackerlogic.c ot_urlencoded.c ot_batch.c ot_buf.c ot_event.c ot_fdstate.c ot_fmt.c ot_hex.c ot_ip.c ot_config.c ot_scan.c ot_socket.c ot_tcp.c ot_mutex.c ot_random.c ot_stats.c ot_vector.c ot_clean.c ot_udp.c ot_iovec.c ot_fullscrape.c ot_accesslist.c ot_http.c ot_livesync.c ot_rijndael.c
 SOURCES_proxy=proxy.c ot_batch.c ot_fdstate.c ot_vector.c ot_mutex.c ot_random.c ot_scan.c ot_ip.c ot_socket.c
 
 all: $(BINARY) $(BINARY).debug
diff --git a/opentracker.c b/opentracker.c
index b47759e..2eaf988 100644
--- a/opentracker.c
+++ b/opentracker.c
@@ -35,6 +35,7 @@
 #include "ot_fdstate.h"
 #include "ot_socket.h"
 #include "ot_stats.h"
+#include "ot_tcp.h"
 #include "ot_udp.h"
 #include "trackerlogic.h"
 
@@ -96,142 +97,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_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 */
-static void handle_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);
-
-    handle_read(sock, ws);
-
-    /* Gone already if it was answered and closed inline */
-    if (fdstate_generation((int)sock) != generation)
-      continue;
-
-    if (http_adopt(sock, &cookie) == -1)
-      http_close(sock);
-    else
-      fdstate_release((int)sock);
-  }
-}
-
 /* How many ready descriptors one wait hands back at a time */
 #define OT_EVENT_BATCH 512
 
@@ -277,7 +142,7 @@ 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)
@@ -285,10 +150,10 @@ static void *server_mainloop(void *args) {
           ;
       else {
         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);
       }
     }
 
diff --git a/ot_tcp.c b/ot_tcp.c
new file mode 100644
index 0000000..8647a47
--- /dev/null
+++ b/ot_tcp.c
@@ -0,0 +1,157 @@
+/* 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 <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;
+
+    if (http_adopt(sock, &cookie) == -1)
+      http_close(sock);
+    else
+      fdstate_release((int)sock);
+  }
+}
diff --git a/ot_tcp.h b/ot_tcp.h
new file mode 100644
index 0000000..b0a93f4
--- /dev/null
+++ b/ot_tcp.h
@@ -0,0 +1,22 @@
+/* This software was written by Dirk Engling <erdgeist@erdgeist.org>
+   It is considered beerware. Prost. Skol. Cheers or whatever.
+
+   $id$ */
+
+#ifndef OT_TCP_H__
+#define OT_TCP_H__
+
+#include <stdint.h>
+
+#include "trackerlogic.h"
+
+/* Take whatever connections are waiting on a listening socket and answer
+   them, each on this thread's stack, handing a connection that cannot be
+   finished to the event loop */
+void tcp_accept(const int64_t serversocket, struct ot_workstruct *ws);
+
+/* Handler connections in the event loop has become readable or writable */
+void tcp_read(const int64_t sock, struct ot_workstruct *ws);
+void tcp_write(const int64_t sock);
+
+#endif