ot_livesync.c

   1 /* This software was written by Dirk Engling <erdgeist@erdgeist.org>
   2  It is considered beerware. Prost. Skol. Cheers or whatever.
   3
   4  $id$ */
   5
   6 /* System */
   7 #include <pthread.h>
   8 #include <stdlib.h>
   9 #include <string.h>
  10 #include <sys/types.h>
  11 #include <sys/uio.h>
  12 #include <unistd.h>
  13
  14 /* Libowfat */
  15 #include "byte.h"
  16 #include "ip6.h"
  17 #include "ndelay.h"
  18 #include "socket.h"
  19
  20 /* Opentracker */
  21 #include "ot_accesslist.h"
  22 #include "ot_livesync.h"
  23 #include "ot_mutex.h"
  24 #include "ot_stats.h"
  25 #include "trackerlogic.h"
  26
  27 #ifdef WANT_SYNC_LIVE
  28
  29 char groupip_1[4] = {224, 0, 23, 5};
  30
  31 #define LIVESYNC_INCOMING_BUFFSIZE        (256 * 256)
  32
  33 #define LIVESYNC_OUTGOING_BUFFSIZE_PEERS  1480
  34 #define LIVESYNC_OUTGOING_WATERMARK_PEERS (sizeof(ot_peer) + sizeof(ot_hash))
  35
  36 #define LIVESYNC_MAXDELAY                 15 /* seconds */
  37
  38 enum { OT_SYNC_PEER4, OT_SYNC_PEER6 };
  39
  40 /* Forward declaration */
  41 static void           *livesync_worker(void *args);
  42
  43 /* For outgoing packets */
  44 static int64           g_socket_in    = -1;
  45
  46 /* For incoming packets */
  47 static int64           g_socket_out   = -1;
  48
  49 static pthread_mutex_t g_outbuf_mutex = PTHREAD_MUTEX_INITIALIZER;
  50 typedef struct {
  51   uint8_t data[LIVESYNC_OUTGOING_BUFFSIZE_PEERS];
  52   size_t  fill;
  53   ot_time next_packet_time;
  54 } sync_buffer;
  55
  56 static sync_buffer g_v6_buf;
  57 static sync_buffer g_v4_buf;
  58
  59 static pthread_t   thread_id;
  60 void livesync_init() {
  61
  62   if (g_socket_in == -1)
  63     exerr("No socket address for live sync specified.");
  64
  65   /* Prepare outgoing peers buffer */
  66   memcpy(g_v6_buf.data, &g_tracker_id, sizeof(g_tracker_id));
  67   memcpy(g_v4_buf.data, &g_tracker_id, sizeof(g_tracker_id));
  68
  69   uint32_pack_big((char *)g_v6_buf.data + sizeof(g_tracker_id), OT_SYNC_PEER6);
  70   uint32_pack_big((char *)g_v4_buf.data + sizeof(g_tracker_id), OT_SYNC_PEER4);
  71
  72   g_v6_buf.fill             = sizeof(g_tracker_id) + sizeof(uint32_t);
  73   g_v4_buf.fill             = sizeof(g_tracker_id) + sizeof(uint32_t);
  74
  75   g_v6_buf.next_packet_time = g_now_seconds + LIVESYNC_MAXDELAY;
  76   g_v4_buf.next_packet_time = g_now_seconds + LIVESYNC_MAXDELAY;
  77
  78   pthread_create(&thread_id, NULL, livesync_worker, NULL);
  79 }
  80
  81 void livesync_deinit() {
  82   if (g_socket_in != -1)
  83     close(g_socket_in);
  84   if (g_socket_out != -1)
  85     close(g_socket_out);
  86
  87   pthread_cancel(thread_id);
  88 }
  89
  90 void livesync_bind_mcast(ot_ip6 ip, uint16_t port) {
  91   char  tmpip[4] = {0, 0, 0, 0};
  92   char *v4ip;
  93
  94   if (!ip6_isv4mapped(ip))
  95     exerr("v6 mcast support not yet available.");
  96   v4ip = ip + 12;
  97
  98   if (g_socket_in != -1)
  99     exerr("Error: Livesync listen ip specified twice.");
 100
 101   if ((g_socket_in = socket_udp4()) < 0)
 102     exerr("Error: Cant create live sync incoming socket.");
 103   ndelay_off(g_socket_in);
 104
 105   if (socket_bind4_reuse(g_socket_in, tmpip, port) == -1)
 106     exerr("Error: Cant bind live sync incoming socket.");
 107
 108   if (socket_mcjoin4(g_socket_in, groupip_1, v4ip))
 109     exerr("Error: Cant make live sync incoming socket join mcast group.");
 110
 111   if ((g_socket_out = socket_udp4()) < 0)
 112     exerr("Error: Cant create live sync outgoing socket.");
 113   if (socket_bind4_reuse(g_socket_out, v4ip, port) == -1)
 114     exerr("Error: Cant bind live sync outgoing socket.");
 115
 116   socket_mcttl4(g_socket_out, 1);
 117   socket_mcloop4(g_socket_out, 0);
 118 }
 119
 120 /* Caller MUST hold g_outbuf_mutex. Returns with g_outbuf_mutex unlocked */
 121 static void livesync_issue_peersync(sync_buffer *buf) {
 122   char   mycopy[LIVESYNC_OUTGOING_BUFFSIZE_PEERS];
 123   size_t fill = buf->fill;
 124
 125   memcpy(mycopy, buf->data, fill);
 126   buf->fill             = sizeof(g_tracker_id) + sizeof(uint32_t);
 127   buf->next_packet_time = g_now_seconds + LIVESYNC_MAXDELAY;
 128
 129   /* From now this thread has a local copy of the buffer and
 130      has modified the protected element */
 131   pthread_mutex_unlock(&g_outbuf_mutex);
 132
 133   socket_send4(g_socket_out, mycopy, fill, groupip_1, LIVESYNC_PORT);
 134 }
 135
 136 static void livesync_handle_peersync(struct ot_workstruct *ws, size_t peer_size) {
 137   size_t off = sizeof(g_tracker_id) + sizeof(uint32_t);
 138
 139   /* Now basic sanity checks have been done on the live sync packet
 140      We might add more testing and logging. */
 141   while ((ssize_t)(off + sizeof(ot_hash) + peer_size) <= ws->request_size) {
 142     memcpy(&ws->peer, ws->request + off + sizeof(ot_hash), peer_size);
 143     ws->hash = (ot_hash *)(ws->request + off);
 144
 145     if (!g_opentracker_running)
 146       return;
 147
 148     if (OT_PEERFLAG(ws->peer) & PEER_FLAG_STOPPED)
 149       remove_peer_from_torrent(FLAG_MCA, ws);
 150     else
 151       add_peer_to_torrent_and_return_peers(FLAG_MCA, ws, /* amount = */ 0);
 152
 153     off += sizeof(ot_hash) + peer_size;
 154   }
 155
 156   stats_issue_event(EVENT_SYNC, 0, (ws->request_size - sizeof(g_tracker_id) - sizeof(uint32_t)) / ((ssize_t)sizeof(ot_hash) + peer_size));
 157 }
 158
 159 /* Tickle the live sync module from time to time, so no events get
 160    stuck when there's not enough traffic to fill udp packets fast
 161    enough */
 162 void livesync_ticker() {
 163   /* livesync_issue_peersync sets g_next_packet_time */
 164   pthread_mutex_lock(&g_outbuf_mutex);
 165   if (g_now_seconds > g_v6_buf.next_packet_time && g_v6_buf.fill > sizeof(g_tracker_id) + sizeof(uint32_t))
 166     livesync_issue_peersync(&g_v6_buf);
 167   else
 168     pthread_mutex_unlock(&g_outbuf_mutex);
 169
 170   pthread_mutex_lock(&g_outbuf_mutex);
 171   if (g_now_seconds > g_v4_buf.next_packet_time && g_v4_buf.fill > sizeof(g_tracker_id) + sizeof(uint32_t))
 172     livesync_issue_peersync(&g_v4_buf);
 173   else
 174     pthread_mutex_unlock(&g_outbuf_mutex);
 175 }
 176
 177 /* Inform live sync about whats going on. */
 178 void livesync_tell(struct ot_workstruct *ws) {
 179   size_t       peer_size; /* initialized in next line */
 180   ot_peer     *peer_src = peer_from_peer6(&ws->peer, &peer_size);
 181   sync_buffer *dest_buf = peer_size == OT_PEER_SIZE6 ? &g_v6_buf : &g_v4_buf;
 182
 183   pthread_mutex_lock(&g_outbuf_mutex);
 184
 185   memcpy(dest_buf->data + dest_buf->fill, ws->hash, sizeof(ot_hash));
 186   dest_buf->fill += sizeof(ot_hash);
 187
 188   memcpy(dest_buf->data + dest_buf->fill, peer_src, peer_size);
 189   dest_buf->fill += peer_size;
 190
 191   if (dest_buf->fill >= LIVESYNC_OUTGOING_BUFFSIZE_PEERS - LIVESYNC_OUTGOING_WATERMARK_PEERS)
 192     livesync_issue_peersync(dest_buf);
 193   else
 194     pthread_mutex_unlock(&g_outbuf_mutex);
 195 }
 196
 197 static void *livesync_worker(void *args) {
 198   struct ot_workstruct ws;
 199   ot_ip6               in_ip;
 200   uint16_t             in_port;
 201
 202   (void)args;
 203
 204   /* Initialize our "thread local storage" */
 205   ws.inbuf = ws.request = malloc(LIVESYNC_INCOMING_BUFFSIZE);
 206   ws.outbuf = ws.reply = 0;
 207
 208   memcpy(in_ip, V4mappedprefix, sizeof(V4mappedprefix));
 209
 210   while (1) {
 211     ws.request_size = socket_recv4(g_socket_in, (char *)ws.inbuf, LIVESYNC_INCOMING_BUFFSIZE, 12 + (char *)in_ip, &in_port);
 212
 213     /* Expect at least tracker id and packet type */
 214     if (ws.request_size <= (ssize_t)(sizeof(g_tracker_id) + sizeof(uint32_t)))
 215       continue;
 216     if (!accesslist_is_blessed(in_ip, OT_PERMISSION_MAY_LIVESYNC))
 217       continue;
 218     if (!memcmp(ws.inbuf, &g_tracker_id, sizeof(g_tracker_id))) {
 219       /* TODO: log packet coming from ourselves */
 220       continue;
 221     }
 222
 223     switch (uint32_read_big(sizeof(g_tracker_id) + (char *)ws.inbuf)) {
 224     case OT_SYNC_PEER6:
 225       livesync_handle_peersync(&ws, OT_PEER_SIZE6);
 226       break;
 227     case OT_SYNC_PEER4:
 228       livesync_handle_peersync(&ws, OT_PEER_SIZE4);
 229       break;
 230     default:
 231       break;
 232     }
 233   }
 234
 235   /* Never returns. */
 236   return NULL;
 237 }
 238
 239 #endif