opentracker/proxy.c

/* This software was written by Dirk Engling <erdgeist@erdgeist.org>
   It is considered beerware. Prost. Skol. Cheers or whatever.

   $Id$ */

/* System */
#include <arpa/inet.h>
#include <ctype.h>
#include <errno.h>
#include <pthread.h>
#include <pwd.h>
#include <signal.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <unistd.h>

/* Libowfat */
#include "byte.h"
#include "io.h"
#include "iob.h"
#include "ip6.h"
#include "ndelay.h"
#include "scan.h"
#include "socket.h"

/* Opentracker */
#include "ot_mutex.h"
#include "ot_stats.h"
#include "ot_vector.h"
#include "trackerlogic.h"

#ifndef WANT_SYNC_LIVE
#define WANT_SYNC_LIVE
#endif
#include "ot_livesync.h"

ot_ip6   g_serverip;
uint16_t g_serverport = 9009;
uint32_t g_tracker_id;
char     groupip_1[4] = {224, 0, 23, 5};
int      g_self_pipe[2];

/* If you have more than 10 peers, don't use this proxy
   Use 20 slots for 10 peers to have room for 10 incoming connection slots
 */
#define MAX_PEERS                         20

#define LIVESYNC_INCOMING_BUFFSIZE        (256 * 256)
#define STREAMSYNC_OUTGOING_BUFFSIZE      (256 * 256)

#define LIVESYNC_OUTGOING_BUFFSIZE_PEERS  1480
#define LIVESYNC_OUTGOING_WATERMARK_PEERS (sizeof(ot_peer) + sizeof(ot_hash))
#define LIVESYNC_MAXDELAY                 15 /* seconds */

/* The amount of time a complete sync cycle should take */
#define OT_SYNC_INTERVAL_MINUTES          2

/* So after each bucket wait 1 / OT_BUCKET_COUNT intervals */
#define OT_SYNC_SLEEP                     (((OT_SYNC_INTERVAL_MINUTES) * 60 * 1000000) / (OT_BUCKET_COUNT))

enum { OT_SYNC_PEER4, OT_SYNC_PEER6 };
enum { FLAG_SERVERSOCKET = 1 };

/* For incoming packets */
static int64    g_socket_in = -1;
static uint8_t  g_inbuffer[LIVESYNC_INCOMING_BUFFSIZE];

/* For outgoing packets */
static int64    g_socket_out = -1;
static uint8_t  g_peerbuffer_start[LIVESYNC_OUTGOING_BUFFSIZE_PEERS];
static uint8_t *g_peerbuffer_pos;
static uint8_t *g_peerbuffer_highwater = g_peerbuffer_start + LIVESYNC_OUTGOING_BUFFSIZE_PEERS - LIVESYNC_OUTGOING_WATERMARK_PEERS;
static ot_time  g_next_packet_time;

static void *livesync_worker(void *args);
static void *streamsync_worker(void *args);
static void livesync_proxytell(uint8_t prefix, uint8_t *info_hash, uint8_t *peer);

void exerr(char *message) {
  fprintf(stderr, "%s\n", message);
  exit(111);
}

void stats_issue_event(ot_status_event event, PROTO_FLAG proto, uintptr_t event_data) {
  (void)event;
  (void)proto;
  (void)event_data;
}

void livesync_bind_mcast(ot_ip6 ip, uint16_t port) {
  char  tmpip[4] = {0, 0, 0, 0};
  char *v4ip;

  if (!ip6_isv4mapped(ip))
    exerr("v6 mcast support not yet available.");
  v4ip = ip + 12;

  if (g_socket_in != -1)
    exerr("Error: Livesync listen ip specified twice.");

  if ((g_socket_in = socket_udp4()) < 0)
    exerr("Error: Cant create live sync incoming socket.");
  ndelay_off(g_socket_in);

  if (socket_bind4_reuse(g_socket_in, tmpip, port) == -1)
    exerr("Error: Cant bind live sync incoming socket.");

  if (socket_mcjoin4(g_socket_in, groupip_1, v4ip))
    exerr("Error: Cant make live sync incoming socket join mcast group.");

  if ((g_socket_out = socket_udp4()) < 0)
    exerr("Error: Cant create live sync outgoing socket.");
  if (socket_bind4_reuse(g_socket_out, v4ip, port) == -1)
    exerr("Error: Cant bind live sync outgoing socket.");

  socket_mcttl4(g_socket_out, 1);
  socket_mcloop4(g_socket_out, 1);
}

size_t add_peer_to_torrent_proxy(ot_hash hash, ot_peer *peer, size_t peer_size) {
  int          exactmatch;
  ot_torrent  *torrent;
  ot_peerlist *peer_list;
  ot_peer     *peer_dest;
  ot_vector   *torrents_list = mutex_bucket_lock_by_hash(hash);
  size_t       compare_size  = OT_PEER_COMPARE_SIZE_FROM_PEER_SIZE(peer_size);

  torrent                    = vector_find_or_insert(torrents_list, (void *)hash, sizeof(ot_torrent), compare_size, &exactmatch);
  if (!torrent)
    return -1;

  if (!exactmatch) {
    /* Create a new torrent entry, then */
    memcpy(torrent->hash, hash, sizeof(ot_hash));

    if (!(torrent->peer_list6 = malloc(sizeof(ot_peerlist))) || !(torrent->peer_list4 = malloc(sizeof(ot_peerlist)))) {
      vector_remove_torrent(torrents_list, torrent);
      mutex_bucket_unlock_by_hash(hash, 0);
      return -1;
    }

    byte_zero(torrent->peer_list6, sizeof(ot_peerlist));
    byte_zero(torrent->peer_list4, sizeof(ot_peerlist));
  }

  peer_list = peer_size == OT_PEER_SIZE6 ? torrent->peer_list6 : torrent->peer_list4;

  /* Check for peer in torrent */
  peer_dest = vector_find_or_insert_peer(&(peer_list->peers), peer, peer_size, &exactmatch);
  if (!peer_dest) {
    mutex_bucket_unlock_by_hash(hash, 0);
    return -1;
  }
  /* Tell peer that it's fresh */
  OT_PEERTIME(peer, peer_size) = 0;

  /* If we hadn't had a match create peer there */
  if (!exactmatch) {
    peer_list->peer_count++;
    if (OT_PEERFLAG_D(peer, peer_size) & PEER_FLAG_SEEDING)
      peer_list->seed_count++;
  }
  memcpy(peer_dest, peer, peer_size);
  mutex_bucket_unlock_by_hash(hash, 0);
  return 0;
}

size_t remove_peer_from_torrent_proxy(ot_hash hash, ot_peer *peer, size_t peer_size) {
  int         exactmatch;
  ot_vector  *torrents_list = mutex_bucket_lock_by_hash(hash);
  ot_torrent *torrent       = binary_search(hash, torrents_list->data, torrents_list->size, sizeof(ot_torrent), OT_HASH_COMPARE_SIZE, &exactmatch);

  if (exactmatch) {
    ot_peerlist *peer_list = peer_list = peer_size == OT_PEER_SIZE6 ? torrent->peer_list6 : torrent->peer_list4;
    switch (vector_remove_peer(&peer_list->peers, peer, peer_size)) {
    case 2:
      peer_list->seed_count--; /* Intentional fallthrough */
    case 1:
      peer_list->peer_count--; /* Intentional fallthrough */
    default:
      break;
    }
  }

  mutex_bucket_unlock_by_hash(hash, 0);
  return 0;
}

void free_peerlist(ot_peerlist *peer_list) {
  if (peer_list->peers.data) {
    if (OT_PEERLIST_HASBUCKETS(peer_list)) {
      ot_vector *bucket_list = (ot_vector *)(peer_list->peers.data);

      while (peer_list->peers.size--)
        free(bucket_list++->data);
    }
    free(peer_list->peers.data);
  }
  free(peer_list);
}

static void livesync_handle_peersync(ssize_t datalen, size_t peer_size) {
  int off = sizeof(g_tracker_id) + sizeof(uint32_t);

  fprintf(stderr, ".");

  while ((ssize_t)(off + sizeof(ot_hash) + peer_size) <= datalen) {
    ot_peer *peer = (ot_peer *)(g_inbuffer + off + sizeof(ot_hash));
    ot_hash *hash = (ot_hash *)(g_inbuffer + off);

    if (OT_PEERFLAG_D(peer, peer_size) & PEER_FLAG_STOPPED)
      remove_peer_from_torrent_proxy(*hash, peer, peer_size);
    else
      add_peer_to_torrent_proxy(*hash, peer, peer_size);

    off += sizeof(ot_hash) + peer_size;
  }
}

int usage(char *self) {
  fprintf(stderr, "Usage: %s -L <livesync_iface_ip> -l <listenip>:<listenport> -c <connectip>:<connectport>\n", self);
  return 0;
}

enum {
  FLAG_OUTGOING      = 0x80,

  FLAG_DISCONNECTED  = 0x00,
  FLAG_CONNECTING    = 0x01,
  FLAG_WAITTRACKERID = 0x02,
  FLAG_CONNECTED     = 0x03,

  FLAG_MASK          = 0x07
};

#define PROXYPEER_NEEDSCONNECT(flag)     ((flag) == FLAG_OUTGOING)
#define PROXYPEER_ISCONNECTED(flag)      (((flag) & FLAG_MASK) == FLAG_CONNECTED)
#define PROXYPEER_SETDISCONNECTED(flag)  (flag) = (((flag) & FLAG_OUTGOING) | FLAG_DISCONNECTED)
#define PROXYPEER_SETCONNECTING(flag)    (flag) = (((flag) & FLAG_OUTGOING) | FLAG_CONNECTING)
#define PROXYPEER_SETWAITTRACKERID(flag) (flag) = (((flag) & FLAG_OUTGOING) | FLAG_WAITTRACKERID)
#define PROXYPEER_SETCONNECTED(flag)     (flag) = (((flag) & FLAG_OUTGOING) | FLAG_CONNECTED)

typedef struct {
  int      state;             /* Whether we want to connect, how far our handshake is, etc. */
  ot_ip6   ip;                /* The peer to connect to */
  uint16_t port;              /* The peers port */
  uint8_t  indata[8192 * 16]; /* Any data not processed yet */
  size_t   indata_length;     /* Length of unprocessed data */
  uint32_t tracker_id;        /* How the other end greeted */
  int64    fd;                /* A file handle, if connected, <= 0 is disconnected (0 initially, -1 else) */
  io_batch outdata;           /* The iobatch containing our sync data */

  size_t   packet_tcount;     /* Number of unprocessed torrents in packet we currently receive */
  uint8_t  packet_tprefix;    /* Prefix byte for all torrents in current packet */
  uint8_t  packet_type;       /* Type of current packet */
  uint32_t packet_tid;        /* Tracker id for current packet */

} proxy_peer;
static void process_indata(proxy_peer *peer);

void reset_info_block(proxy_peer *peer) {
  peer->indata_length = 0;
  peer->tracker_id    = 0;
  peer->fd            = -1;
  peer->packet_tcount = 0;
  iob_reset(&peer->outdata);
  PROXYPEER_SETDISCONNECTED(peer->state);
}

/* Number of connections to peers
 * If a peer's IP is set, we try to reconnect, when the connection drops
 * If we already have a connected tracker_id in our records for an _incoming_ connection, drop it
 * Multiple connections to/from the same ip are okay, if tracker_id doesn't match
 * Reconnect attempts occur only twice a minute
 */
static int        g_connection_count;
static ot_time    g_connection_reconn;
static proxy_peer g_connections[MAX_PEERS];

static void handle_reconnects(void) {
  int i;
  for (i = 0; i < g_connection_count; ++i)
    if (PROXYPEER_NEEDSCONNECT(g_connections[i].state)) {
      int64 newfd = socket_tcp6();
      fprintf(stderr, "(Re)connecting to peer...");
      if (newfd < 0)
        continue; /* No socket for you */
      io_fd(newfd);
      if (socket_bind6_reuse(newfd, g_serverip, g_serverport, 0)) {
        io_close(newfd);
        continue;
      }
      if (socket_connect6(newfd, g_connections[i].ip, g_connections[i].port, 0) == -1 && errno != EINPROGRESS && errno != EWOULDBLOCK) {
        close(newfd);
        continue;
      }
      io_wantwrite(newfd); /* So we will be informed when it is connected */
      io_setcookie(newfd, g_connections + i);

      /* Prepare connection info block */
      reset_info_block(g_connections + i);
      g_connections[i].fd = newfd;
      PROXYPEER_SETCONNECTING(g_connections[i].state);
    }
  g_connection_reconn = time(NULL) + 30;
}

/* Handle incoming connection requests, check against whitelist */
static void handle_accept(int64 serversocket) {
  int64  newfd;
  ot_ip6 ip;
  uint16 port;

  while ((newfd = socket_accept6(serversocket, ip, &port, NULL)) != -1) {

    /* XXX some access control */

    /* Put fd into a non-blocking mode */
    io_nonblock(newfd);

    if (!io_fd(newfd))
      io_close(newfd);
    else {
      /* Find a new home for our incoming connection */
      int i;
      for (i = 0; i < MAX_PEERS; ++i)
        if (g_connections[i].state == FLAG_DISCONNECTED)
          break;
      if (i == MAX_PEERS) {
        fprintf(stderr, "No room for incoming connection.");
        close(newfd);
        continue;
      }

      /* Prepare connection info block */
      reset_info_block(g_connections + i);
      PROXYPEER_SETCONNECTING(g_connections[i].state);
      g_connections[i].port = port;
      g_connections[i].fd   = newfd;

      io_setcookie(newfd, g_connections + i);

      /* We expect the connecting side to begin with its tracker_id */
      io_wantread(newfd);
    }
  }

  return;
}

/* New sync data on the stream */
static void handle_read(int64 peersocket) {
  int         i;
  int64       datalen;
  uint32_t    tracker_id;
  proxy_peer *peer = io_getcookie(peersocket);

  if (!peer) {
    /* Can't happen ;) */
    io_close(peersocket);
    return;
  }
  switch (peer->state & FLAG_MASK) {
  case FLAG_DISCONNECTED:
    io_close(peersocket);
    break; /* Shouldnt happen */
  case FLAG_CONNECTING:
  case FLAG_WAITTRACKERID:
    /* We want at least the first four bytes to come at once, to avoid keeping extra states (for now)
       This also catches 0 bytes reads == EOF and negative values, denoting connection errors */
    if (io_tryread(peersocket, (void *)&tracker_id, sizeof(tracker_id)) != sizeof(tracker_id))
      goto close_socket;

    /* See, if we already have a connection to that peer */
    for (i = 0; i < MAX_PEERS; ++i)
      if ((g_connections[i].state & FLAG_MASK) == FLAG_CONNECTED && g_connections[i].tracker_id == tracker_id) {
        fprintf(stderr, "Peer already connected. Closing connection.\n");
        goto close_socket;
      }

    /* Also no need for soliloquy */
    if (tracker_id == g_tracker_id)
      goto close_socket;

    /* The new connection is good, send our tracker_id on incoming connections */
    if (peer->state == FLAG_CONNECTING)
      if (io_trywrite(peersocket, (void *)&g_tracker_id, sizeof(g_tracker_id)) != sizeof(g_tracker_id))
        goto close_socket;

    peer->tracker_id = tracker_id;
    PROXYPEER_SETCONNECTED(peer->state);

    if (peer->state & FLAG_OUTGOING)
      fprintf(stderr, "succeeded.\n");
    else
      fprintf(stderr, "Incoming connection successful.\n");

    break;
  close_socket:
    fprintf(stderr, "Handshake incomplete, closing socket\n");
    io_close(peersocket);
    reset_info_block(peer);
    break;
  case FLAG_CONNECTED:
    /* Here we acutally expect data from peer
       indata_length should be less than 20+256*7 bytes, for incomplete torrent entries */
    datalen = io_tryread(peersocket, (void *)(peer->indata + peer->indata_length), sizeof(peer->indata) - peer->indata_length);
    if (!datalen || datalen < -1) {
      fprintf(stderr, "Connection closed by remote peer.\n");
      io_close(peersocket);
      reset_info_block(peer);
    } else if (datalen > 0) {
      peer->indata_length += datalen;
      process_indata(peer);
    }
    break;
  }
}

/* Can write new sync data to the stream */
static void handle_write(int64 peersocket) {
  proxy_peer *peer = io_getcookie(peersocket);

  if (!peer) {
    /* Can't happen ;) */
    io_close(peersocket);
    return;
  }

  switch (peer->state & FLAG_MASK) {
  case FLAG_DISCONNECTED:
  default: /* Should not happen */
    io_close(peersocket);
    break;
  case FLAG_CONNECTING:
    /* Ensure that the connection is established and handle connection error */
    if (peer->state & FLAG_OUTGOING && !socket_connected(peersocket)) {
      fprintf(stderr, "failed\n");
      reset_info_block(peer);
      io_close(peersocket);
      break;
    }

    if (io_trywrite(peersocket, (void *)&g_tracker_id, sizeof(g_tracker_id)) == sizeof(g_tracker_id)) {
      PROXYPEER_SETWAITTRACKERID(peer->state);
      io_dontwantwrite(peersocket);
      io_wantread(peersocket);
    } else {
      fprintf(stderr, "Handshake incomplete, closing socket\n");
      io_close(peersocket);
      reset_info_block(peer);
    }
    break;
  case FLAG_CONNECTED:
    switch (iob_send(peersocket, &peer->outdata)) {
    case 0: /* all data sent */
      io_dontwantwrite(peersocket);
      break;
    case -3: /* an error occured */
      io_close(peersocket);
      reset_info_block(peer);
      break;
    default: /* Normal operation or eagain */
      break;
    }
    break;
  }

  return;
}

static void server_mainloop() {
  int64 sock;

  /* inlined livesync_init() */
  memset(g_peerbuffer_start, 0, sizeof(g_peerbuffer_start));
  g_peerbuffer_pos = g_peerbuffer_start;
  memcpy(g_peerbuffer_pos, &g_tracker_id, sizeof(g_tracker_id));
  uint32_pack_big((char *)g_peerbuffer_pos + sizeof(g_tracker_id), OT_SYNC_PEER);
  g_peerbuffer_pos   += sizeof(g_tracker_id) + sizeof(uint32_t);
  g_next_packet_time  = time(NULL) + LIVESYNC_MAXDELAY;

  while (1) {
    /* See if we need to connect to anyone */
    if (time(NULL) > g_connection_reconn)
      handle_reconnects();

    /* Wait for io events until next approx reconn check time */
    io_waituntil2(30 * 1000);

    /* Loop over readable sockets */
    while ((sock = io_canread()) != -1) {
      const void *cookie = io_getcookie(sock);
      if ((uintptr_t)cookie == FLAG_SERVERSOCKET)
        handle_accept(sock);
      else
        handle_read(sock);
    }

    /* Loop over writable sockets */
    while ((sock = io_canwrite()) != -1)
      handle_write(sock);

    livesync_ticker();
  }
}

static void panic(const char *routine) {
  fprintf(stderr, "%s: %s\n", routine, strerror(errno));
  exit(111);
}

static int64_t ot_try_bind(ot_ip6 ip, uint16_t port) {
  int64 sock = socket_tcp6();

  if (socket_bind6_reuse(sock, ip, port, 0) == -1)
    panic("socket_bind6_reuse");

  if (socket_listen(sock, SOMAXCONN) == -1)
    panic("socket_listen");

  if (!io_fd(sock))
    panic("io_fd");

  io_setcookie(sock, (void *)FLAG_SERVERSOCKET);
  io_wantread(sock);
  return sock;
}

static int scan_ip6_port(const char *src, ot_ip6 ip, uint16 *port) {
  const char *s = src;
  int         off, bracket = 0;
  while (isspace(*s))
    ++s;
  if (*s == '[')
    ++s, ++bracket; /* for v6 style notation */
  if (!(off = scan_ip6(s, ip)))
    return 0;
  s += off;
  if (*s == 0 || isspace(*s))
    return s - src;
  if (*s == ']' && bracket)
    ++s;
  if (!ip6_isv4mapped(ip)) {
    if ((bracket && *(s) != ':') || (*(s) != '.'))
      return 0;
    s++;
  } else {
    if (*(s++) != ':')
      return 0;
  }
  if (!(off = scan_ushort(s, port)))
    return 0;
  return off + s - src;
}

int main(int argc, char **argv) {
  static pthread_t sync_in_thread_id;
  static pthread_t sync_out_thread_id;
  ot_ip6           serverip;
  uint16_t         tmpport;
  int              scanon = 1, lbound = 0, sbound = 0;

  srandom(time(NULL));
#ifdef WANT_ARC4RANDOM
  g_tracker_id = arc4random();
#else
  g_tracker_id = random();
#endif

  while (scanon) {
    switch (getopt(argc, argv, ":l:c:L:h")) {
    case -1:
      scanon = 0;
      break;
    case 'l':
      tmpport = 0;
      if (!scan_ip6_port(optarg, serverip, &tmpport) || !tmpport) {
        usage(argv[0]);
        exit(1);
      }
      ot_try_bind(serverip, tmpport);
      ++sbound;
      break;
    case 'c':
      if (g_connection_count > MAX_PEERS / 2)
        exerr("Connection limit exceeded.\n");
      tmpport = 0;
      if (!scan_ip6_port(optarg, g_connections[g_connection_count].ip, &g_connections[g_connection_count].port) || !g_connections[g_connection_count].port) {
        usage(argv[0]);
        exit(1);
      }
      g_connections[g_connection_count++].state = FLAG_OUTGOING;
      break;
    case 'L':
      tmpport = 9696;
      if (!scan_ip6_port(optarg, serverip, &tmpport) || !tmpport) {
        usage(argv[0]);
        exit(1);
      }
      livesync_bind_mcast(serverip, tmpport);
      ++lbound;
      break;
    default:
    case '?':
      usage(argv[0]);
      exit(1);
    }
  }

  if (!lbound)
    exerr("No livesync port bound.");
  if (!g_connection_count && !sbound)
    exerr("No streamsync port bound.");
  pthread_create(&sync_in_thread_id, NULL, livesync_worker, NULL);
  pthread_create(&sync_out_thread_id, NULL, streamsync_worker, NULL);

  server_mainloop();
  return 0;
}

static void *streamsync_worker(void *args) {
  (void)args;
  while (1) {
    int bucket;
    /* For each bucket... */
    for (bucket = 0; bucket < OT_BUCKET_COUNT; ++bucket) {
      /* Get exclusive access to that bucket */
      ot_vector *torrents_list = mutex_bucket_lock(bucket);
      size_t     tor_offset, count_def = 0, count_one = 0, count_two = 0, count_peers = 0;
      size_t     mem, mem_a = 0, mem_b = 0;
      uint8_t   *ptr = 0, *ptr_a, *ptr_b, *ptr_c;

      if (!torrents_list->size)
        goto unlock_continue;

      /* For each torrent in this bucket.. */
      for (tor_offset = 0; tor_offset < torrents_list->size; ++tor_offset) {
        /* Address torrents members */
        ot_peerlist *peer_list = (((ot_torrent *)(torrents_list->data))[tor_offset]).peer_list;
        switch (peer_list->peer_count) {
        case 2:
          count_two++;
          break;
        case 1:
          count_one++;
          break;
        case 0:
          break;
        default:
          count_def++;
          count_peers += peer_list->peer_count;
        }
      }

      /* Maximal memory requirement: max 3 blocks, max torrents * 20 + max peers * 7 */
      mem = 3 * (1 + 1 + 2) + (count_one + count_two) * (19 + 1) + count_def * (19 + 8) + (count_one + 2 * count_two + count_peers) * 7;

      fprintf(stderr, "Mem: %zd\n", mem);

      ptr = ptr_a = ptr_b = ptr_c = malloc(mem);
      if (!ptr)
        goto unlock_continue;

      if (count_one > 4 || !count_def) {
        mem_a     = 1 + 1 + 2 + count_one * (19 + 7);
        ptr_b    += mem_a;
        ptr_c    += mem_a;
        ptr_a[0]  = 1;                                     /* Offset 0: packet type 1 */
        ptr_a[1]  = (bucket << 8) >> OT_BUCKET_COUNT_BITS; /* Offset 1: the shared prefix */
        ptr_a[2]  = count_one >> 8;
        ptr_a[3]  = count_one & 255;
        ptr_a    += 4;
      } else
        count_def += count_one;

      if (count_two > 4 || !count_def) {
        mem_b     = 1 + 1 + 2 + count_two * (19 + 14);
        ptr_c    += mem_b;
        ptr_b[0]  = 2;                                     /* Offset 0: packet type 2 */
        ptr_b[1]  = (bucket << 8) >> OT_BUCKET_COUNT_BITS; /* Offset 1: the shared prefix */
        ptr_b[2]  = count_two >> 8;
        ptr_b[3]  = count_two & 255;
        ptr_b    += 4;
      } else
        count_def += count_two;

      if (count_def) {
        ptr_c[0]  = 0;                                     /* Offset 0: packet type 0 */
        ptr_c[1]  = (bucket << 8) >> OT_BUCKET_COUNT_BITS; /* Offset 1: the shared prefix */
        ptr_c[2]  = count_def >> 8;
        ptr_c[3]  = count_def & 255;
        ptr_c    += 4;
      }

      /* For each torrent in this bucket.. */
      for (tor_offset = 0; tor_offset < torrents_list->size; ++tor_offset) {
        /* Address torrents members */
        ot_torrent  *torrent   = ((ot_torrent *)(torrents_list->data)) + tor_offset;
        ot_peerlist *peer_list = torrent->peer_list;
        ot_peer     *peers     = (ot_peer *)(peer_list->peers.data);
        uint8_t    **dst;

        /* Determine destination slot */
        count_peers = peer_list->peer_count;
        switch (count_peers) {
        case 0:
          continue;
        case 1:
          dst = mem_a ? &ptr_a : &ptr_c;
          break;
        case 2:
          dst = mem_b ? &ptr_b : &ptr_c;
          break;
        default:
          dst = &ptr_c;
          break;
        }

        /* Copy tail of info_hash, advance pointer */
        memcpy(*dst, ((uint8_t *)torrent->hash) + 1, sizeof(ot_hash) - 1);
        *dst += sizeof(ot_hash) - 1;

        /* Encode peer count */
        if (dst == &ptr_c)
          while (count_peers) {
            if (count_peers <= 0x7f)
              *(*dst)++ = count_peers;
            else
              *(*dst)++ = 0x80 | (count_peers & 0x7f);
            count_peers >>= 7;
          }

        /* Copy peers */
        count_peers = peer_list->peer_count;
        while (count_peers--) {
          memcpy(*dst, peers++, OT_IP_SIZE + 3);
          *dst += OT_IP_SIZE + 3;
        }
        free_peerlist(peer_list);
      }

      free(torrents_list->data);
      memset(torrents_list, 0, sizeof(*torrents_list));
    unlock_continue:
      mutex_bucket_unlock(bucket, 0);

      if (ptr) {
        int i;

        if (ptr_b > ptr_c)
          ptr_c = ptr_b;
        if (ptr_a > ptr_c)
          ptr_c = ptr_a;
        mem = ptr_c - ptr;

        for (i = 0; i < MAX_PEERS; ++i) {
          if (PROXYPEER_ISCONNECTED(g_connections[i].state)) {
            void *tmp = malloc(mem);
            if (tmp) {
              memcpy(tmp, ptr, mem);
              iob_addbuf_free(&g_connections[i].outdata, tmp, mem);
              io_wantwrite(g_connections[i].fd);
            }
          }
        }

        free(ptr);
      }
      usleep(OT_SYNC_SLEEP);
    }
  }
  return 0;
}

static void livesync_issue_peersync() {
  socket_send4(g_socket_out, (char *)g_peerbuffer_start, g_peerbuffer_pos - g_peerbuffer_start, groupip_1, LIVESYNC_PORT);
  g_peerbuffer_pos   = g_peerbuffer_start + sizeof(g_tracker_id) + sizeof(uint32_t);
  g_next_packet_time = time(NULL) + LIVESYNC_MAXDELAY;
}

void livesync_ticker() {
  /* livesync_issue_peersync sets g_next_packet_time */
  if (time(NULL) > g_next_packet_time && g_peerbuffer_pos > g_peerbuffer_start + sizeof(g_tracker_id))
    livesync_issue_peersync();
}

static void livesync_proxytell(uint8_t prefix, uint8_t *info_hash, uint8_t *peer) {
  //  unsigned int i;

  *g_peerbuffer_pos = prefix;
  memcpy(g_peerbuffer_pos + 1, info_hash, sizeof(ot_hash) - 1);
  memcpy(g_peerbuffer_pos + sizeof(ot_hash), peer, sizeof(ot_peer) - 1);

#if 0
  /* Dump info_hash */
  for( i=0; i<sizeof(ot_hash); ++i )
    printf( "%02X", g_peerbuffer_pos[i] );
  putchar( ':' );
#endif
  g_peerbuffer_pos += sizeof(ot_hash);
#if 0
  printf( "%hhu.%hhu.%hhu.%hhu:%hu (%02X %02X)\n", g_peerbuffer_pos[0], g_peerbuffer_pos[1], g_peerbuffer_pos[2], g_peerbuffer_pos[3],
    g_peerbuffer_pos[4] | ( g_peerbuffer_pos[5] << 8 ), g_peerbuffer_pos[6], g_peerbuffer_pos[7] );
#endif
  g_peerbuffer_pos += sizeof(ot_peer);

  if (g_peerbuffer_pos >= g_peerbuffer_highwater)
    livesync_issue_peersync();
}

static void process_indata(proxy_peer *peer) {
  size_t   consumed, peers;
  uint8_t *data    = peer->indata, *hash;
  uint8_t *dataend = data + peer->indata_length;

  while (1) {
    /* If we're not inside of a packet, make a new one */
    if (!peer->packet_tcount) {
      /* Ensure the header is complete or postpone processing */
      if (data + 4 > dataend)
        break;
      peer->packet_type     = data[0];
      peer->packet_tprefix  = data[1];
      peer->packet_tcount   = data[2] * 256 + data[3];
      data                 += 4;
      printf("type: %hhu, prefix: %02X, torrentcount: %zd\n", peer->packet_type, peer->packet_tprefix, peer->packet_tcount);
    }

    /* Ensure size for a minimal torrent block */
    if (data + sizeof(ot_hash) + OT_IP_SIZE + 3 > dataend)
      break;

    /* Advance pointer to peer count or peers */
    hash   = data;
    data  += sizeof(ot_hash) - 1;

    /* Type 0 has peer count encoded before each peers */
    peers  = peer->packet_type;
    if (!peers) {
      int shift = 0;
      do
        peers |= (0x7f & *data) << (7 * shift);
      while (*(data++) & 0x80 && shift++ < 6);
    }
#if 0
printf( "peers: %zd\n", peers );
#endif
    /* Ensure enough data being read to hold all peers */
    if (data + (OT_IP_SIZE + 3) * peers > dataend) {
      data = hash;
      break;
    }
    while (peers--) {
      livesync_proxytell(peer->packet_tprefix, hash, data);
      data += OT_IP_SIZE + 3;
    }
    --peer->packet_tcount;
  }

  consumed = data - peer->indata;
  memmove(peer->indata, data, peer->indata_length - consumed);
  peer->indata_length -= consumed;
}

static void *livesync_worker(void *args) {
  (void)args;
  while (1) {
    ot_ip6   in_ip;
    uint16_t in_port;
    size_t   datalen = socket_recv4(g_socket_in, (char *)g_inbuffer, LIVESYNC_INCOMING_BUFFSIZE, 12 + (char *)in_ip, &in_port);

    /* Expect at least tracker id and packet type */
    if (datalen <= (ssize_t)(sizeof(g_tracker_id) + sizeof(uint32_t)))
      continue;
    if (!memcmp(g_inbuffer, &g_tracker_id, sizeof(g_tracker_id))) {
      /* drop packet coming from ourselves */
      continue;
    }
    switch (uint32_read_big((char *)g_inbuffer + sizeof(g_tracker_id))) {
    case OT_SYNC_PEER4:
      livesync_handle_peersync(datalen, OT_PEER_SIZE4);
      break;
    case OT_SYNC_PEER6:
      livesync_handle_peersync(datalen, OT_PEER_SIZE6);
      break;
    default:
      // fprintf( stderr, "Received an unknown live sync packet type %u.\n", uint32_read_big( sizeof( g_tracker_id ) + (char*)g_inbuffer ) );
      break;
    }
  }
  return 0;
}