summaryrefslogtreecommitdiff
path: root/src/proxy.c
diff options
context:
space:
mode:
Diffstat (limited to 'src/proxy.c')
-rw-r--r--src/proxy.c871
1 files changed, 871 insertions, 0 deletions
diff --git a/src/proxy.c b/src/proxy.c
new file mode 100644
index 0000000..8589db5
--- /dev/null
+++ b/src/proxy.c
@@ -0,0 +1,871 @@
1/* This software was written by Dirk Engling <erdgeist@erdgeist.org>
2 It is considered beerware. Prost. Skol. Cheers or whatever. */
3
4/* This relays sync traffic between trackers, and between multicast domains
5 that have no route to each other, so that no tracker has to reach out to
6 the outside world by itself. Defunc. */
7
8/* System */
9#include <arpa/inet.h>
10#include <ctype.h>
11#include <errno.h>
12#include <pthread.h>
13#include <stdint.h>
14#include <stdio.h>
15#include <stdlib.h>
16#include <string.h>
17#include <unistd.h>
18
19/* Libowfat */
20#include "io.h"
21
22/* Opentracker */
23#include "ot_batch.h"
24#include "ot_fdstate.h"
25#include "ot_ip.h"
26#include "ot_mutex.h"
27#include "ot_random.h"
28#include "ot_scan.h"
29#include "ot_socket.h"
30#include "ot_stats.h"
31#include "ot_vector.h"
32#include "trackerlogic.h"
33
34#include "ot_livesync.h"
35
36ot_ip6 g_serverip;
37uint16_t g_serverport = 9009;
38uint32_t g_tracker_id;
39ot_ip6 groupip_1 = {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0xff, 0xff, 224, 0, 23, 5};
40int g_self_pipe[2];
41
42/* If you have more than 10 peers, don't use this proxy
43 Use 20 slots for 10 peers to have room for 10 incoming connection slots
44 */
45#define MAX_PEERS 20
46
47#define LIVESYNC_INCOMING_BUFFSIZE (256 * 256)
48#define STREAMSYNC_OUTGOING_BUFFSIZE (256 * 256)
49
50#define LIVESYNC_OUTGOING_BUFFSIZE_PEERS 1480
51#define LIVESYNC_OUTGOING_WATERMARK_PEERS (sizeof(ot_peer) + sizeof(ot_hash))
52#define LIVESYNC_MAXDELAY 15 /* seconds */
53
54/* The amount of time a complete sync cycle should take */
55#define OT_SYNC_INTERVAL_MINUTES 2
56
57/* So after each bucket wait 1 / OT_BUCKET_COUNT intervals */
58#define OT_SYNC_SLEEP (((OT_SYNC_INTERVAL_MINUTES) * 60 * 1000000) / (OT_BUCKET_COUNT))
59
60enum { OT_SYNC_PEER4, OT_SYNC_PEER6 };
61enum { FLAG_SERVERSOCKET = 1 };
62
63/* For incoming packets */
64static int g_socket_in = -1;
65static uint8_t g_inbuffer[LIVESYNC_INCOMING_BUFFSIZE];
66
67/* For outgoing packets */
68static int g_socket_out = -1;
69static uint8_t g_peerbuffer_start[LIVESYNC_OUTGOING_BUFFSIZE_PEERS];
70static uint8_t *g_peerbuffer_pos;
71static uint8_t *g_peerbuffer_highwater = g_peerbuffer_start + LIVESYNC_OUTGOING_BUFFSIZE_PEERS - LIVESYNC_OUTGOING_WATERMARK_PEERS;
72static ot_time g_next_packet_time;
73
74static void *livesync_worker(void *args);
75static void *streamsync_worker(void *args);
76static void livesync_proxytell(uint8_t prefix, uint8_t *info_hash, uint8_t *peer);
77
78void exerr(const char *message) {
79 fprintf(stderr, "%s\n", message);
80 exit(111);
81}
82
83void stats_issue_event(struct ot_workstruct *ws, ot_status_event event, PROTO_FLAG proto, uintptr_t event_data) {
84 (void)ws;
85 (void)event;
86 (void)proto;
87 (void)event_data;
88}
89
90void livesync_bind_mcast(ot_ip6 ip, uint16_t port) {
91 if (!ip_is_v4mapped(ip))
92 exerr("v6 mcast support not yet available.");
93
94 if (g_socket_in != -1)
95 exerr("Error: Livesync listen ip specified twice.");
96
97 if ((g_socket_in = socket_mcast_join(groupip_1, ip, port)) == -1)
98 exerr("Error: Cant bind live sync incoming socket.");
99
100 if ((g_socket_out = socket_mcast_bind(ip, port, 1)) == -1)
101 exerr("Error: Cant bind live sync outgoing socket.");
102}
103
104size_t add_peer_to_torrent_proxy(ot_hash hash, ot_peer *peer, size_t peer_size) {
105 int exactmatch;
106 ot_torrent *torrent;
107 ot_peerlist *peer_list;
108 ot_peer *peer_dest;
109 ot_vector *torrents_list = mutex_bucket_lock_by_hash(hash);
110 size_t compare_size = OT_PEER_COMPARE_SIZE_FROM_PEER_SIZE(peer_size);
111
112 torrent = vector_find_or_insert(torrents_list, (void *)hash, sizeof(ot_torrent), compare_size, &exactmatch);
113 if (!torrent)
114 return -1;
115
116 if (!exactmatch) {
117 /* Create a new torrent entry, then */
118 memcpy(torrent->hash, hash, sizeof(ot_hash));
119
120 if (!(torrent->peer_list6 = malloc(sizeof(ot_peerlist))) || !(torrent->peer_list4 = malloc(sizeof(ot_peerlist)))) {
121 vector_remove_torrent(torrents_list, torrent);
122 mutex_bucket_unlock_by_hash(hash, 0);
123 return -1;
124 }
125
126 memset(torrent->peer_list6, 0, sizeof(ot_peerlist));
127 memset(torrent->peer_list4, 0, sizeof(ot_peerlist));
128 }
129
130 peer_list = peer_size == OT_PEER_SIZE6 ? torrent->peer_list6 : torrent->peer_list4;
131
132 /* Check for peer in torrent */
133 peer_dest = vector_find_or_insert_peer(&(peer_list->peers), peer, peer_size, &exactmatch);
134 if (!peer_dest) {
135 mutex_bucket_unlock_by_hash(hash, 0);
136 return -1;
137 }
138 /* Tell peer that it's fresh */
139 OT_PEERTIME(peer, peer_size) = 0;
140
141 /* If we hadn't had a match create peer there */
142 if (!exactmatch) {
143 peer_list->peer_count++;
144 if (OT_PEERFLAG_D(peer, peer_size) & PEER_FLAG_SEEDING)
145 peer_list->seed_count++;
146 }
147 memcpy(peer_dest, peer, peer_size);
148 mutex_bucket_unlock_by_hash(hash, 0);
149 return 0;
150}
151
152size_t remove_peer_from_torrent_proxy(ot_hash hash, ot_peer *peer, size_t peer_size) {
153 int exactmatch;
154 ot_vector *torrents_list = mutex_bucket_lock_by_hash(hash);
155 ot_torrent *torrent = vector_binary_search(hash, torrents_list->data, torrents_list->size, sizeof(ot_torrent), OT_HASH_COMPARE_SIZE, &exactmatch);
156
157 if (exactmatch) {
158 ot_peerlist *peer_list = peer_list = peer_size == OT_PEER_SIZE6 ? torrent->peer_list6 : torrent->peer_list4;
159 switch (vector_remove_peer(&peer_list->peers, peer, peer_size)) {
160 case 2:
161 peer_list->seed_count--; /* Intentional fallthrough */
162 case 1:
163 peer_list->peer_count--; /* Intentional fallthrough */
164 default:
165 break;
166 }
167 }
168
169 mutex_bucket_unlock_by_hash(hash, 0);
170 return 0;
171}
172
173void free_peerlist(ot_peerlist *peer_list) {
174 if (peer_list->peers.data) {
175 if (OT_PEERLIST_HASBUCKETS(peer_list)) {
176 ot_vector *bucket_list = (ot_vector *)(peer_list->peers.data);
177
178 while (peer_list->peers.size--)
179 free(bucket_list++->data);
180 }
181 free(peer_list->peers.data);
182 }
183 free(peer_list);
184}
185
186static void livesync_handle_peersync(ssize_t datalen, size_t peer_size) {
187 int off = sizeof(g_tracker_id) + sizeof(uint32_t);
188
189 fprintf(stderr, ".");
190
191 while ((ssize_t)(off + sizeof(ot_hash) + peer_size) <= datalen) {
192 ot_peer *peer = (ot_peer *)(g_inbuffer + off + sizeof(ot_hash));
193 ot_hash *hash = (ot_hash *)(g_inbuffer + off);
194
195 if (OT_PEERFLAG_D(peer, peer_size) & PEER_FLAG_STOPPED)
196 remove_peer_from_torrent_proxy(*hash, peer, peer_size);
197 else
198 add_peer_to_torrent_proxy(*hash, peer, peer_size);
199
200 off += sizeof(ot_hash) + peer_size;
201 }
202}
203
204int usage(char *self) {
205 fprintf(stderr, "Usage: %s -L <livesync_iface_ip> -l <listenip>:<listenport> -c <connectip>:<connectport>\n", self);
206 return 0;
207}
208
209enum {
210 FLAG_OUTGOING = 0x80,
211
212 FLAG_DISCONNECTED = 0x00,
213 FLAG_CONNECTING = 0x01,
214 FLAG_WAITTRACKERID = 0x02,
215 FLAG_CONNECTED = 0x03,
216
217 FLAG_MASK = 0x07
218};
219
220#define PROXYPEER_NEEDSCONNECT(flag) ((flag) == FLAG_OUTGOING)
221#define PROXYPEER_ISCONNECTED(flag) (((flag) & FLAG_MASK) == FLAG_CONNECTED)
222#define PROXYPEER_SETDISCONNECTED(flag) (flag) = (((flag) & FLAG_OUTGOING) | FLAG_DISCONNECTED)
223#define PROXYPEER_SETCONNECTING(flag) (flag) = (((flag) & FLAG_OUTGOING) | FLAG_CONNECTING)
224#define PROXYPEER_SETWAITTRACKERID(flag) (flag) = (((flag) & FLAG_OUTGOING) | FLAG_WAITTRACKERID)
225#define PROXYPEER_SETCONNECTED(flag) (flag) = (((flag) & FLAG_OUTGOING) | FLAG_CONNECTED)
226
227typedef struct {
228 int state; /* Whether we want to connect, how far our handshake is, etc. */
229 ot_ip6 ip; /* The peer to connect to */
230 uint16_t port; /* The peers port */
231 uint8_t indata[8192 * 16]; /* Any data not processed yet */
232 size_t indata_length; /* Length of unprocessed data */
233 uint32_t tracker_id; /* How the other end greeted */
234 int64_t fd; /* A file handle, if connected, <= 0 is disconnected (0 initially, -1 else) */
235 ot_batch outdata; /* The batch containing our sync data */
236
237 size_t packet_tcount; /* Number of unprocessed torrents in packet we currently receive */
238 uint8_t packet_tprefix; /* Prefix byte for all torrents in current packet */
239 uint8_t packet_type; /* Type of current packet */
240 uint32_t packet_tid; /* Tracker id for current packet */
241
242} proxy_peer;
243static void process_indata(proxy_peer *peer);
244
245void reset_info_block(proxy_peer *peer) {
246 peer->indata_length = 0;
247 peer->tracker_id = 0;
248 peer->fd = -1;
249 peer->packet_tcount = 0;
250 batch_reset(&peer->outdata);
251 PROXYPEER_SETDISCONNECTED(peer->state);
252}
253
254/* Number of connections to peers
255 * If a peer's IP is set, we try to reconnect, when the connection drops
256 * If we already have a connected tracker_id in our records for an _incoming_ connection, drop it
257 * Multiple connections to/from the same ip are okay, if tracker_id doesn't match
258 * Reconnect attempts occur only twice a minute
259 */
260static size_t g_connection_count;
261static ot_time g_connection_reconn;
262static proxy_peer g_connections[MAX_PEERS];
263
264static void handle_reconnects(void) {
265 size_t i;
266 for (i = 0; i < g_connection_count; ++i)
267 if (PROXYPEER_NEEDSCONNECT(g_connections[i].state)) {
268 int64_t newfd = socket_tcp_connect(g_serverip, g_serverport, g_connections[i].ip, g_connections[i].port);
269 fprintf(stderr, "(Re)connecting to peer...");
270 if (newfd == -1)
271 continue; /* No socket for you */
272 io_fd(newfd);
273 io_wantwrite(newfd); /* So we will be informed when it is connected */
274 fdstate_open(newfd, g_connections + i, 0);
275
276 /* Prepare connection info block */
277 reset_info_block(g_connections + i);
278 g_connections[i].fd = newfd;
279 PROXYPEER_SETCONNECTING(g_connections[i].state);
280 }
281 g_connection_reconn = time(NULL) + 30;
282}
283
284/* Handle incoming connection requests, check against whitelist */
285static void handle_accept(int64_t serversocket) {
286 int64_t newfd;
287 ot_ip6 ip;
288 uint16_t port;
289
290 while ((newfd = socket_accept(serversocket, ip, &port)) != -1) {
291
292 /* XXX some access control */
293
294 /* Put fd into a non-blocking mode */
295 io_nonblock(newfd);
296
297 if (!io_fd(newfd))
298 io_close(newfd);
299 else {
300 /* Find a new home for our incoming connection */
301 int i;
302 for (i = 0; i < MAX_PEERS; ++i)
303 if (g_connections[i].state == FLAG_DISCONNECTED)
304 break;
305 if (i == MAX_PEERS) {
306 fprintf(stderr, "No room for incoming connection.");
307 close(newfd);
308 continue;
309 }
310
311 /* Prepare connection info block */
312 reset_info_block(g_connections + i);
313 PROXYPEER_SETCONNECTING(g_connections[i].state);
314 g_connections[i].port = port;
315 g_connections[i].fd = newfd;
316
317 fdstate_open(newfd, g_connections + i, 0);
318
319 /* We expect the connecting side to begin with its tracker_id */
320 io_wantread(newfd);
321 }
322 }
323
324 return;
325}
326
327/* New sync data on the stream */
328static void handle_read(int64_t peersocket) {
329 int i;
330 int64_t datalen;
331 uint32_t tracker_id;
332 proxy_peer *peer = fdstate_cookie(peersocket);
333
334 if (!peer) {
335 /* Can't happen ;) */
336 io_close(peersocket);
337 return;
338 }
339 switch (peer->state & FLAG_MASK) {
340 case FLAG_DISCONNECTED:
341 io_close(peersocket);
342 break; /* Shouldnt happen */
343 case FLAG_CONNECTING:
344 case FLAG_WAITTRACKERID:
345 /* We want at least the first four bytes to come at once, to avoid keeping extra states (for now)
346 This also catches 0 bytes reads == EOF and negative values, denoting connection errors */
347 if (io_tryread(peersocket, (void *)&tracker_id, sizeof(tracker_id)) != sizeof(tracker_id))
348 goto close_socket;
349
350 /* See, if we already have a connection to that peer */
351 for (i = 0; i < MAX_PEERS; ++i)
352 if ((g_connections[i].state & FLAG_MASK) == FLAG_CONNECTED && g_connections[i].tracker_id == tracker_id) {
353 fprintf(stderr, "Peer already connected. Closing connection.\n");
354 goto close_socket;
355 }
356
357 /* Also no need for soliloquy */
358 if (tracker_id == g_tracker_id)
359 goto close_socket;
360
361 /* The new connection is good, send our tracker_id on incoming connections */
362 if (peer->state == FLAG_CONNECTING)
363 if (io_trywrite(peersocket, (void *)&g_tracker_id, sizeof(g_tracker_id)) != sizeof(g_tracker_id))
364 goto close_socket;
365
366 peer->tracker_id = tracker_id;
367 PROXYPEER_SETCONNECTED(peer->state);
368
369 if (peer->state & FLAG_OUTGOING)
370 fprintf(stderr, "succeeded.\n");
371 else
372 fprintf(stderr, "Incoming connection successful.\n");
373
374 break;
375 close_socket:
376 fprintf(stderr, "Handshake incomplete, closing socket\n");
377 io_close(peersocket);
378 reset_info_block(peer);
379 break;
380 case FLAG_CONNECTED:
381 /* Here we acutally expect data from peer
382 indata_length should be less than 20+256*7 bytes, for incomplete torrent entries */
383 datalen = io_tryread(peersocket, (void *)(peer->indata + peer->indata_length), sizeof(peer->indata) - peer->indata_length);
384 if (!datalen || datalen < -1) {
385 fprintf(stderr, "Connection closed by remote peer.\n");
386 io_close(peersocket);
387 reset_info_block(peer);
388 } else if (datalen > 0) {
389 peer->indata_length += datalen;
390 process_indata(peer);
391 }
392 break;
393 }
394}
395
396/* Can write new sync data to the stream */
397static void handle_write(int64_t peersocket) {
398 proxy_peer *peer = fdstate_cookie(peersocket);
399
400 if (!peer) {
401 /* Can't happen ;) */
402 io_close(peersocket);
403 return;
404 }
405
406 switch (peer->state & FLAG_MASK) {
407 case FLAG_DISCONNECTED:
408 default: /* Should not happen */
409 io_close(peersocket);
410 break;
411 case FLAG_CONNECTING:
412 /* Ensure that the connection is established and handle connection error */
413 if (peer->state & FLAG_OUTGOING && !socket_is_connected(peersocket)) {
414 fprintf(stderr, "failed\n");
415 reset_info_block(peer);
416 io_close(peersocket);
417 break;
418 }
419
420 if (io_trywrite(peersocket, (void *)&g_tracker_id, sizeof(g_tracker_id)) == sizeof(g_tracker_id)) {
421 PROXYPEER_SETWAITTRACKERID(peer->state);
422 io_dontwantwrite(peersocket);
423 io_wantread(peersocket);
424 } else {
425 fprintf(stderr, "Handshake incomplete, closing socket\n");
426 io_close(peersocket);
427 reset_info_block(peer);
428 }
429 break;
430 case FLAG_CONNECTED:
431 switch (batch_send(peersocket, &peer->outdata)) {
432 case BATCH_DONE:
433 io_dontwantwrite(peersocket);
434 break;
435 case BATCH_FAILED:
436 io_close(peersocket);
437 reset_info_block(peer);
438 break;
439 default: /* Normal operation or eagain */
440 break;
441 }
442 break;
443 }
444
445 return;
446}
447
448static void server_mainloop() {
449 int64_t sock;
450 uint32_t packet_type;
451
452 /* inlined livesync_init() */
453 memset(g_peerbuffer_start, 0, sizeof(g_peerbuffer_start));
454 g_peerbuffer_pos = g_peerbuffer_start;
455 memcpy(g_peerbuffer_pos, &g_tracker_id, sizeof(g_tracker_id));
456 packet_type = htonl(OT_SYNC_PEER);
457 memcpy(g_peerbuffer_pos + sizeof(g_tracker_id), &packet_type, sizeof(packet_type));
458 g_peerbuffer_pos += sizeof(g_tracker_id) + sizeof(uint32_t);
459 g_next_packet_time = time(NULL) + LIVESYNC_MAXDELAY;
460
461 while (1) {
462 /* See if we need to connect to anyone */
463 if (time(NULL) > g_connection_reconn)
464 handle_reconnects();
465
466 /* Wait for io events until next approx reconn check time */
467 io_waituntil2(30 * 1000);
468
469 /* Loop over readable sockets */
470 while ((sock = io_canread()) != -1) {
471 const void *cookie = fdstate_cookie(sock);
472 if ((uintptr_t)cookie == FLAG_SERVERSOCKET)
473 handle_accept(sock);
474 else
475 handle_read(sock);
476 }
477
478 /* Loop over writable sockets */
479 while ((sock = io_canwrite()) != -1)
480 handle_write(sock);
481
482 livesync_ticker();
483 }
484}
485
486static void panic(const char *routine) {
487 fprintf(stderr, "%s: %s\n", routine, strerror(errno));
488 exit(111);
489}
490
491static int64_t ot_try_bind(ot_ip6 ip, uint16_t port) {
492 int64_t sock = socket_tcp_listen(ip, port, SOCKET_EXCLUSIVE);
493
494 if (sock == -1)
495 panic("socket_tcp_listen");
496
497 if (!io_fd(sock))
498 panic("io_fd");
499
500 fdstate_open(sock, (void *)FLAG_SERVERSOCKET, 0);
501 io_wantread(sock);
502 return sock;
503}
504
505static int scan_ip6_port(const char *src, ot_ip6 ip, uint16_t *port) {
506 const char *s = src;
507 uint64_t value;
508 int off, bracket = 0;
509 while (isspace(*s))
510 ++s;
511 if (*s == '[')
512 ++s, ++bracket; /* for v6 style notation */
513 if (!(off = ip_parse(s, ip)))
514 return 0;
515 s += off;
516 if (*s == 0 || isspace(*s))
517 return s - src;
518 if (*s == ']' && bracket)
519 ++s;
520 if (!ip_is_v4mapped(ip)) {
521 if ((bracket && *(s) != ':') || (*(s) != '.'))
522 return 0;
523 s++;
524 } else {
525 if (*(s++) != ':')
526 return 0;
527 }
528 if (!(off = scan_uint(s, &value, 65535)))
529 return 0;
530 *port = (uint16_t)value;
531 return off + s - src;
532}
533
534int main(int argc, char **argv) {
535 static pthread_t sync_in_thread_id;
536 static pthread_t sync_out_thread_id;
537 ot_ip6 serverip;
538 uint16_t tmpport;
539 int scanon = 1, lbound = 0, sbound = 0;
540
541 random_init();
542 if (fdstate_init() == -1)
543 panic("fdstate_init");
544 g_tracker_id = random_uint32();
545
546 while (scanon) {
547 switch (getopt(argc, argv, ":l:c:L:h")) {
548 case -1:
549 scanon = 0;
550 break;
551 case 'l':
552 tmpport = 0;
553 if (!scan_ip6_port(optarg, serverip, &tmpport) || !tmpport) {
554 usage(argv[0]);
555 exit(1);
556 }
557 ot_try_bind(serverip, tmpport);
558 ++sbound;
559 break;
560 case 'c':
561 if (g_connection_count > MAX_PEERS / 2)
562 exerr("Connection limit exceeded.\n");
563 tmpport = 0;
564 if (!scan_ip6_port(optarg, g_connections[g_connection_count].ip, &g_connections[g_connection_count].port) || !g_connections[g_connection_count].port) {
565 usage(argv[0]);
566 exit(1);
567 }
568 g_connections[g_connection_count++].state = FLAG_OUTGOING;
569 break;
570 case 'L':
571 tmpport = 9696;
572 if (!scan_ip6_port(optarg, serverip, &tmpport) || !tmpport) {
573 usage(argv[0]);
574 exit(1);
575 }
576 livesync_bind_mcast(serverip, tmpport);
577 ++lbound;
578 break;
579 default:
580 case '?':
581 usage(argv[0]);
582 exit(1);
583 }
584 }
585
586 if (!lbound)
587 exerr("No livesync port bound.");
588 if (!g_connection_count && !sbound)
589 exerr("No streamsync port bound.");
590 pthread_create(&sync_in_thread_id, NULL, livesync_worker, NULL);
591 pthread_create(&sync_out_thread_id, NULL, streamsync_worker, NULL);
592
593 server_mainloop();
594 return 0;
595}
596
597static void *streamsync_worker(void *args) {
598 (void)args;
599 while (1) {
600 size_t bucket;
601 /* For each bucket... */
602 for (bucket = 0; bucket < OT_BUCKET_COUNT; ++bucket) {
603 /* Get exclusive access to that bucket */
604 ot_vector *torrents_list = mutex_bucket_lock(bucket);
605 size_t tor_offset, count_def = 0, count_one = 0, count_two = 0, count_peers = 0;
606 size_t mem, mem_a = 0, mem_b = 0;
607 uint8_t *ptr = 0, *ptr_a, *ptr_b, *ptr_c;
608
609 if (!torrents_list->size)
610 goto unlock_continue;
611
612 /* For each torrent in this bucket.. */
613 for (tor_offset = 0; tor_offset < torrents_list->size; ++tor_offset) {
614 /* Address torrents members */
615 ot_peerlist *peer_list = (((ot_torrent *)(torrents_list->data))[tor_offset]).peer_list;
616 switch (peer_list->peer_count) {
617 case 2:
618 count_two++;
619 break;
620 case 1:
621 count_one++;
622 break;
623 case 0:
624 break;
625 default:
626 count_def++;
627 count_peers += peer_list->peer_count;
628 }
629 }
630
631 /* Maximal memory requirement: max 3 blocks, max torrents * 20 + max peers * 7 */
632 mem = 3 * (1 + 1 + 2) + (count_one + count_two) * (19 + 1) + count_def * (19 + 8) + (count_one + 2 * count_two + count_peers) * 7;
633
634 fprintf(stderr, "Mem: %zd\n", mem);
635
636 ptr = ptr_a = ptr_b = ptr_c = malloc(mem);
637 if (!ptr)
638 goto unlock_continue;
639
640 if (count_one > 4 || !count_def) {
641 mem_a = 1 + 1 + 2 + count_one * (19 + 7);
642 ptr_b += mem_a;
643 ptr_c += mem_a;
644 ptr_a[0] = 1; /* Offset 0: packet type 1 */
645 ptr_a[1] = (bucket << 8) >> OT_BUCKET_COUNT_BITS; /* Offset 1: the shared prefix */
646 ptr_a[2] = count_one >> 8;
647 ptr_a[3] = count_one & 255;
648 ptr_a += 4;
649 } else
650 count_def += count_one;
651
652 if (count_two > 4 || !count_def) {
653 mem_b = 1 + 1 + 2 + count_two * (19 + 14);
654 ptr_c += mem_b;
655 ptr_b[0] = 2; /* Offset 0: packet type 2 */
656 ptr_b[1] = (bucket << 8) >> OT_BUCKET_COUNT_BITS; /* Offset 1: the shared prefix */
657 ptr_b[2] = count_two >> 8;
658 ptr_b[3] = count_two & 255;
659 ptr_b += 4;
660 } else
661 count_def += count_two;
662
663 if (count_def) {
664 ptr_c[0] = 0; /* Offset 0: packet type 0 */
665 ptr_c[1] = (bucket << 8) >> OT_BUCKET_COUNT_BITS; /* Offset 1: the shared prefix */
666 ptr_c[2] = count_def >> 8;
667 ptr_c[3] = count_def & 255;
668 ptr_c += 4;
669 }
670
671 /* For each torrent in this bucket.. */
672 for (tor_offset = 0; tor_offset < torrents_list->size; ++tor_offset) {
673 /* Address torrents members */
674 ot_torrent *torrent = ((ot_torrent *)(torrents_list->data)) + tor_offset;
675 ot_peerlist *peer_list = torrent->peer_list;
676 ot_peer *peers = (ot_peer *)(peer_list->peers.data);
677 uint8_t **dst;
678
679 /* Determine destination slot */
680 count_peers = peer_list->peer_count;
681 switch (count_peers) {
682 case 0:
683 continue;
684 case 1:
685 dst = mem_a ? &ptr_a : &ptr_c;
686 break;
687 case 2:
688 dst = mem_b ? &ptr_b : &ptr_c;
689 break;
690 default:
691 dst = &ptr_c;
692 break;
693 }
694
695 /* Copy tail of info_hash, advance pointer */
696 memcpy(*dst, ((uint8_t *)torrent->hash) + 1, sizeof(ot_hash) - 1);
697 *dst += sizeof(ot_hash) - 1;
698
699 /* Encode peer count */
700 if (dst == &ptr_c)
701 while (count_peers) {
702 if (count_peers <= 0x7f)
703 *(*dst)++ = count_peers;
704 else
705 *(*dst)++ = 0x80 | (count_peers & 0x7f);
706 count_peers >>= 7;
707 }
708
709 /* Copy peers */
710 count_peers = peer_list->peer_count;
711 while (count_peers--) {
712 memcpy(*dst, peers++, OT_IP_SIZE + 3);
713 *dst += OT_IP_SIZE + 3;
714 }
715 free_peerlist(peer_list);
716 }
717
718 free(torrents_list->data);
719 memset(torrents_list, 0, sizeof(*torrents_list));
720 unlock_continue:
721 mutex_bucket_unlock(bucket, 0);
722
723 if (ptr) {
724 int i;
725
726 if (ptr_b > ptr_c)
727 ptr_c = ptr_b;
728 if (ptr_a > ptr_c)
729 ptr_c = ptr_a;
730 mem = ptr_c - ptr;
731
732 for (i = 0; i < MAX_PEERS; ++i) {
733 if (PROXYPEER_ISCONNECTED(g_connections[i].state)) {
734 void *tmp = malloc(mem);
735 if (tmp) {
736 memcpy(tmp, ptr, mem);
737 batch_add(&g_connections[i].outdata, tmp, mem);
738 io_wantwrite(g_connections[i].fd);
739 }
740 }
741 }
742
743 free(ptr);
744 }
745 usleep(OT_SYNC_SLEEP);
746 }
747 }
748 return 0;
749}
750
751static void livesync_issue_peersync() {
752 socket_mcast_send(g_socket_out, g_peerbuffer_start, (size_t)(g_peerbuffer_pos - g_peerbuffer_start), groupip_1, LIVESYNC_PORT);
753 g_peerbuffer_pos = g_peerbuffer_start + sizeof(g_tracker_id) + sizeof(uint32_t);
754 g_next_packet_time = time(NULL) + LIVESYNC_MAXDELAY;
755}
756
757void livesync_ticker() {
758 /* livesync_issue_peersync sets g_next_packet_time */
759 if (time(NULL) > g_next_packet_time && g_peerbuffer_pos > g_peerbuffer_start + sizeof(g_tracker_id))
760 livesync_issue_peersync();
761}
762
763static void livesync_proxytell(uint8_t prefix, uint8_t *info_hash, uint8_t *peer) {
764 // unsigned int i;
765
766 *g_peerbuffer_pos = prefix;
767 memcpy(g_peerbuffer_pos + 1, info_hash, sizeof(ot_hash) - 1);
768 memcpy(g_peerbuffer_pos + sizeof(ot_hash), peer, sizeof(ot_peer) - 1);
769
770#if 0
771 /* Dump info_hash */
772 for( i=0; i<sizeof(ot_hash); ++i )
773 printf( "%02X", g_peerbuffer_pos[i] );
774 putchar( ':' );
775#endif
776 g_peerbuffer_pos += sizeof(ot_hash);
777#if 0
778 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],
779 g_peerbuffer_pos[4] | ( g_peerbuffer_pos[5] << 8 ), g_peerbuffer_pos[6], g_peerbuffer_pos[7] );
780#endif
781 g_peerbuffer_pos += sizeof(ot_peer);
782
783 if (g_peerbuffer_pos >= g_peerbuffer_highwater)
784 livesync_issue_peersync();
785}
786
787static void process_indata(proxy_peer *peer) {
788 size_t consumed, peers;
789 uint8_t *data = peer->indata, *hash;
790 uint8_t *dataend = data + peer->indata_length;
791
792 while (1) {
793 /* If we're not inside of a packet, make a new one */
794 if (!peer->packet_tcount) {
795 /* Ensure the header is complete or postpone processing */
796 if (data + 4 > dataend)
797 break;
798 peer->packet_type = data[0];
799 peer->packet_tprefix = data[1];
800 peer->packet_tcount = data[2] * 256 + data[3];
801 data += 4;
802 printf("type: %hhu, prefix: %02X, torrentcount: %zd\n", peer->packet_type, peer->packet_tprefix, peer->packet_tcount);
803 }
804
805 /* Ensure size for a minimal torrent block */
806 if (data + sizeof(ot_hash) + OT_IP_SIZE + 3 > dataend)
807 break;
808
809 /* Advance pointer to peer count or peers */
810 hash = data;
811 data += sizeof(ot_hash) - 1;
812
813 /* Type 0 has peer count encoded before each peers */
814 peers = peer->packet_type;
815 if (!peers) {
816 int shift = 0;
817 do
818 peers |= (0x7f & *data) << (7 * shift);
819 while (*(data++) & 0x80 && shift++ < 6);
820 }
821#if 0
822printf( "peers: %zd\n", peers );
823#endif
824 /* Ensure enough data being read to hold all peers */
825 if (data + (OT_IP_SIZE + 3) * peers > dataend) {
826 data = hash;
827 break;
828 }
829 while (peers--) {
830 livesync_proxytell(peer->packet_tprefix, hash, data);
831 data += OT_IP_SIZE + 3;
832 }
833 --peer->packet_tcount;
834 }
835
836 consumed = data - peer->indata;
837 memmove(peer->indata, data, peer->indata_length - consumed);
838 peer->indata_length -= consumed;
839}
840
841static void *livesync_worker(void *args) {
842 (void)args;
843 while (1) {
844 ot_ip6 in_ip;
845 uint16_t in_port;
846 uint32_t packet_type;
847 ssize_t datalen = socket_recv(g_socket_in, g_inbuffer, LIVESYNC_INCOMING_BUFFSIZE, in_ip, &in_port);
848
849 /* Expect at least tracker id and packet type */
850 if (datalen <= (ssize_t)(sizeof(g_tracker_id) + sizeof(uint32_t)))
851 continue;
852 if (!memcmp(g_inbuffer, &g_tracker_id, sizeof(g_tracker_id))) {
853 /* drop packet coming from ourselves */
854 continue;
855 }
856 memcpy(&packet_type, g_inbuffer + sizeof(g_tracker_id), sizeof(packet_type));
857
858 switch (ntohl(packet_type)) {
859 case OT_SYNC_PEER4:
860 livesync_handle_peersync(datalen, OT_PEER_SIZE4);
861 break;
862 case OT_SYNC_PEER6:
863 livesync_handle_peersync(datalen, OT_PEER_SIZE6);
864 break;
865 default:
866 // fprintf( stderr, "Received an unknown live sync packet type %u.\n", ntohl( packet_type ) );
867 break;
868 }
869 }
870 return 0;
871}
generated by cgit v1.3 (git 2.53.0) at 2026-11-02 23:37:18 +0000