path: root/ot_tcp.c

/* 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;
}