/* An HTTP load generator speaking the tracker's own protocol.

   Each request is its own connection, which is what a BitTorrent client
   normally does and what the tracker's hot path is built for: accept, read
   one request, write a short reply, close. That makes this a measurement of
   connection cost rather than of request cost, and connection cost is where
   the tracker's system calls are.

     cc -O2 -pthread -o http_load tests/http_load.c
     ./http_load 127.0.0.1 6969 8 5 10000

   Arguments: host, port, threads, seconds, and how many distinct torrents to
   announce. Cycling a fixed set reaches a steady state where almost every
   announce renews a peer that is already there, rather than measuring how
   fast the tracker can insert torrents it will never see again. */

#include <arpa/inet.h>
#include <errno.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <pthread.h>
#include <stdatomic.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <sys/time.h>
#include <unistd.h>

static struct sockaddr_in g_tracker;
static long               g_torrents = 10000;
static double             g_seconds  = 5;
static _Atomic long long  g_replies  = 0;
static _Atomic long long  g_failed   = 0;
static _Atomic int        g_running  = 1;

static double             now_seconds(void) {
  struct timeval tv;
  gettimeofday(&tv, NULL);
  return tv.tv_sec + tv.tv_usec / 1000000.0;
}

/* %xx for all twenty bytes, which is what a client sends and what the
   tracker's urlencode reader has to walk */
static void hash_of(long which, char *out) {
  int i;

  for (i = 0; i < 20; ++i) {
    unsigned byte = (unsigned)((which >> ((i % 4) * 8)) & 0xff) ^ (unsigned)(i * 37);
    sprintf(out + i * 3, "%%%02x", byte & 0xff);
  }
}

static void *worker(void *args) {
  long  index = (long)(intptr_t)args;
  long  which = index * 7919;
  char  request[512], hash[61], reply[2048];

  while (atomic_load_explicit(&g_running, memory_order_relaxed)) {
    int     sock = socket(AF_INET, SOCK_STREAM, 0);
    int     one  = 1;
    ssize_t got, sent;
    size_t  length;

    if (sock < 0) {
      atomic_fetch_add_explicit(&g_failed, 1, memory_order_relaxed);
      continue;
    }
    setsockopt(sock, IPPROTO_TCP, TCP_NODELAY, &one, sizeof(one));

    if (connect(sock, (struct sockaddr *)&g_tracker, sizeof(g_tracker)) < 0) {
      atomic_fetch_add_explicit(&g_failed, 1, memory_order_relaxed);
      close(sock);
      continue;
    }

    hash_of(which, hash);
    which = (which + 1) % g_torrents;
    length = (size_t)snprintf(request, sizeof(request),
                              "GET /announce?info_hash=%s&peer_id=-OT0001-%06ld%06ld&port=%u&left=0 HTTP/1.0\r\n\r\n",
                              hash, index, which, (unsigned)(6881 + index));

    for (sent = 0; (size_t)sent < length;) {
      ssize_t wrote = write(sock, request + sent, length - (size_t)sent);
      if (wrote <= 0)
        break;
      sent += wrote;
    }

    /* Read to end of reply, which the tracker marks by closing */
    got = 0;
    for (;;) {
      ssize_t piece = read(sock, reply, sizeof(reply));
      if (piece <= 0)
        break;
      got += piece;
    }
    close(sock);

    if (got > 0)
      atomic_fetch_add_explicit(&g_replies, 1, memory_order_relaxed);
    else
      atomic_fetch_add_explicit(&g_failed, 1, memory_order_relaxed);
  }
  return NULL;
}

int main(int argc, char **argv) {
  const char *host    = argc > 1 ? argv[1] : "127.0.0.1";
  int         port    = argc > 2 ? atoi(argv[2]) : 6969;
  long        threads = argc > 3 ? atol(argv[3]) : 4;
  pthread_t  *pool;
  double      started, elapsed;
  long        i;

  g_seconds  = argc > 4 ? atof(argv[4]) : 5;
  g_torrents = argc > 5 ? atol(argv[5]) : 10000;
  if (g_torrents < 1)
    g_torrents = 1;

  memset(&g_tracker, 0, sizeof(g_tracker));
  g_tracker.sin_family = AF_INET;
  g_tracker.sin_port   = htons((uint16_t)port);
  if (inet_pton(AF_INET, host, &g_tracker.sin_addr) != 1) {
    fprintf(stderr, "cannot parse address %s\n", host);
    return 1;
  }

  pool    = calloc((size_t)threads, sizeof(pthread_t));
  started = now_seconds();
  for (i = 0; i < threads; ++i)
    pthread_create(pool + i, NULL, worker, (void *)(intptr_t)i);

  while (now_seconds() - started < g_seconds)
    usleep(20000);
  atomic_store(&g_running, 0);

  for (i = 0; i < threads; ++i)
    pthread_join(pool[i], NULL);
  elapsed = now_seconds() - started;

  printf("%.0f replies/s  (%lld replies, %lld failed, %ld threads, %.1f s)\n",
         (double)atomic_load(&g_replies) / elapsed, (long long)atomic_load(&g_replies),
         (long long)atomic_load(&g_failed), threads, elapsed);
  free(pool);
  return 0;
}