diff options
-rw-r--r-- | proxy.c | 120 |
1 files changed, 77 insertions, 43 deletions
@@ -368,7 +368,7 @@ static void handle_read( int64 peersocket ) { | |||
368 | 368 | ||
369 | /* See, if we already have a connection to that peer */ | 369 | /* See, if we already have a connection to that peer */ |
370 | for( i=0; i<MAX_PEERS; ++i ) | 370 | for( i=0; i<MAX_PEERS; ++i ) |
371 | if( ( g_connections[i].state & FLAG_MASK ) == FLAG_CONNECTED && | 371 | if( ( g_connections[i].state & FLAG_MASK ) == FLAG_CONNECTED && |
372 | g_connections[i].tracker_id == tracker_id ) { | 372 | g_connections[i].tracker_id == tracker_id ) { |
373 | fprintf( stderr, "Peer already connected. Closing connection.\n" ); | 373 | fprintf( stderr, "Peer already connected. Closing connection.\n" ); |
374 | goto close_socket; | 374 | goto close_socket; |
@@ -557,8 +557,8 @@ int main( int argc, char **argv ) { | |||
557 | ot_try_bind( serverip, tmpport ); | 557 | ot_try_bind( serverip, tmpport ); |
558 | ++sbound; | 558 | ++sbound; |
559 | break; | 559 | break; |
560 | case 'c': | 560 | case 'c': |
561 | if( g_connection_count > MAX_PEERS / 2 ) exerr( "Connection limit exceeded.\n" ); | 561 | if( g_connection_count > MAX_PEERS / 2 ) exerr( "Connection limit exceeded.\n" ); |
562 | tmpport = 0; | 562 | tmpport = 0; |
563 | if( !scan_ip6_port( optarg, | 563 | if( !scan_ip6_port( optarg, |
564 | g_connections[g_connection_count].ip, | 564 | g_connections[g_connection_count].ip, |
@@ -613,46 +613,42 @@ static void * streamsync_worker( void * args ) { | |||
613 | } | 613 | } |
614 | 614 | ||
615 | /* Maximal memory requirement: max 3 blocks, max torrents * 20 + max peers * 7 */ | 615 | /* Maximal memory requirement: max 3 blocks, max torrents * 20 + max peers * 7 */ |
616 | mem = 3 * ( 4 + 1 + 1 + 2 ) + ( count_one + count_two ) * 19 + count_def * 20 + | 616 | mem = 3 * ( 4 + 1 + 1 + 2 ) + ( count_one + count_two ) * 19 + count_def * ( 19 + 6 ) + |
617 | ( count_one + 2 * count_two + count_peers ) * 7; | 617 | ( count_one + 2 * count_two + count_peers ) * 7; |
618 | 618 | ||
619 | fprintf( stderr, "Mem: %zd\n", mem ); | 619 | fprintf( stderr, "Mem: %zd\n", mem ); |
620 | 620 | ||
621 | ptr = ptr_a = ptr_b = ptr_c = malloc( mem ); | 621 | ptr = ptr_a = ptr_b = ptr_c = malloc( mem ); |
622 | if( !ptr ) goto unlock_continue; | 622 | if( !ptr ) goto unlock_continue; |
623 | 623 | ||
624 | if( count_one > 8 ) { | 624 | if( count_one > 8 || !count_def ) { |
625 | mem_a = 4 + 1 + 1 + 2 + count_one * ( 19 + 7 ); | 625 | mem_a = 4 + 1 + 1 + 2 + count_one * ( 19 + 7 ); |
626 | ptr_b += mem_a; ptr_c += mem_a; | 626 | ptr_b += mem_a; ptr_c += mem_a; |
627 | memcpy( ptr_a, &g_tracker_id, sizeof(g_tracker_id)); /* Offset 0: the tracker ID */ | 627 | memcpy( ptr_a, &g_tracker_id, sizeof(g_tracker_id)); /* Offset 0: the tracker ID */ |
628 | ptr_a[4] = 1; /* Offset 4: packet type 1 */ | 628 | ptr_a[4] = 1; /* Offset 4: packet type 1 */ |
629 | ptr_a[5] = (bucket << 8) >> OT_BUCKET_COUNT_BITS; /* Offset 5: the shared prefix */ | 629 | ptr_a[5] = (bucket << 8) >> OT_BUCKET_COUNT_BITS; /* Offset 5: the shared prefix */ |
630 | ptr_a[6] = count_one >> 8; | 630 | ptr_a[6] = count_one >> 8; |
631 | ptr_a[7] = count_one & 255; | 631 | ptr_a[7] = count_one & 255; |
632 | ptr_a += 8; | 632 | ptr_a += 8; |
633 | } else { | 633 | } else |
634 | count_def += count_one; | 634 | count_def += count_one; |
635 | count_peers += count_one; | ||
636 | } | ||
637 | 635 | ||
638 | if( count_two > 8 ) { | 636 | if( count_two > 8 || !count_def ) { |
639 | mem_b = 4 + 1 + 1 + 2 + count_two * ( 19 + 14 ); | 637 | mem_b = 4 + 1 + 1 + 2 + count_two * ( 19 + 14 ); |
640 | ptr_c += mem_b; | 638 | ptr_c += mem_b; |
641 | memcpy( ptr_b, &g_tracker_id, sizeof(g_tracker_id)); /* Offset 0: the tracker ID */ | 639 | memcpy( ptr_b, &g_tracker_id, sizeof(g_tracker_id)); /* Offset 0: the tracker ID */ |
642 | ptr_b[4] = 2; /* Offset 4: packet type 2 */ | 640 | ptr_b[4] = 2; /* Offset 4: packet type 2 */ |
643 | ptr_b[5] = (bucket << 8) >> OT_BUCKET_COUNT_BITS; /* Offset 5: the shared prefix */ | 641 | ptr_b[5] = (bucket << 8) >> OT_BUCKET_COUNT_BITS; /* Offset 5: the shared prefix */ |
644 | ptr_b[6] = count_two >> 8; | 642 | ptr_b[6] = count_two >> 8; |
645 | ptr_b[7] = count_two & 255; | 643 | ptr_b[7] = count_two & 255; |
646 | ptr_b += 8; | 644 | ptr_b += 8; |
647 | } else { | 645 | } else |
648 | count_def += count_two; | 646 | count_def += count_two; |
649 | count_peers += 2 * count_two; | ||
650 | } | ||
651 | 647 | ||
652 | if( count_def ) { | 648 | if( count_def ) { |
653 | memcpy( ptr_c, &g_tracker_id, sizeof(g_tracker_id)); /* Offset 0: the tracker ID */ | 649 | memcpy( ptr_c, &g_tracker_id, sizeof(g_tracker_id)); /* Offset 0: the tracker ID */ |
654 | ptr_c[4] = 0; /* Offset 4: packet type 0 */ | 650 | ptr_c[4] = 0; /* Offset 4: packet type 0 */ |
655 | ptr_c[5] = (bucket << 8) >> OT_BUCKET_COUNT_BITS; /* Offset 5: the shared prefix */ | 651 | ptr_c[5] = (bucket << 8) >> OT_BUCKET_COUNT_BITS; /* Offset 5: the shared prefix */ |
656 | ptr_c[6] = count_def >> 8; | 652 | ptr_c[6] = count_def >> 8; |
657 | ptr_c[7] = count_def & 255; | 653 | ptr_c[7] = count_def & 255; |
658 | ptr_c += 8; | 654 | ptr_c += 8; |
@@ -665,26 +661,36 @@ static void * streamsync_worker( void * args ) { | |||
665 | ot_peerlist *peer_list = torrent->peer_list; | 661 | ot_peerlist *peer_list = torrent->peer_list; |
666 | ot_peer *peers = (ot_peer*)(peer_list->peers.data); | 662 | ot_peer *peers = (ot_peer*)(peer_list->peers.data); |
667 | uint8_t **dst; | 663 | uint8_t **dst; |
668 | int multi = 0; | 664 | |
669 | switch( peer_list->peer_count ) { | 665 | /* Determine destination slot */ |
666 | count_peers = peer_list->peer_count; | ||
667 | switch( count_peers ) { | ||
670 | case 0: continue; | 668 | case 0: continue; |
671 | case 1: dst = mem_a ? &ptr_a : &ptr_c; break; | 669 | case 1: dst = mem_a ? &ptr_a : &ptr_c; break; |
672 | case 2: dst = mem_b ? &ptr_b : &ptr_c; break; | 670 | case 2: dst = mem_b ? &ptr_b : &ptr_c; break; |
673 | default: dst = &ptr_c; multi = 1; break; | 671 | default: dst = &ptr_c; break; |
674 | } | 672 | } |
675 | 673 | ||
676 | do { | 674 | /* Copy tail of info_hash, advance pointer */ |
677 | size_t i, pc = peer_list->peer_count; | 675 | memcpy( *dst, torrent->hash + 1, sizeof( ot_hash ) - 1); |
678 | if( pc > 255 ) pc = 255; | 676 | *dst += sizeof( ot_hash ) - 1; |
679 | memcpy( *dst, torrent->hash + 1, sizeof( ot_hash ) - 1); | 677 | |
680 | *dst += sizeof( ot_hash ) - 1; | 678 | /* Encode peer count */ |
681 | if( multi ) *(*dst)++ = pc; | 679 | if( dst == &ptr_c ) |
682 | for( i=0; i < pc; ++i ) { | 680 | while( count_peers ) { |
683 | memcpy( *dst, peers++, OT_IP_SIZE + 3 ); | 681 | if( count_peers <= 0x7f ) |
684 | *dst += OT_IP_SIZE + 3; | 682 | *(*dst)++ = count_peers; |
683 | else | ||
684 | *(*dst)++ = 0x80 | ( count_peers & 0x7f ); | ||
685 | count_peers >>= 7; | ||
685 | } | 686 | } |
686 | peer_list->peer_count -= pc; | 687 | |
687 | } while( peer_list->peer_count ); | 688 | /* Copy peers */ |
689 | count_peers = peer_list->peer_count; | ||
690 | while( count_peers-- ) { | ||
691 | memcpy( *dst, peers++, OT_IP_SIZE + 3 ); | ||
692 | *dst += OT_IP_SIZE + 3; | ||
693 | } | ||
688 | free_peerlist(peer_list); | 694 | free_peerlist(peer_list); |
689 | } | 695 | } |
690 | 696 | ||
@@ -700,7 +706,7 @@ unlock_continue: | |||
700 | if( ptr_a > ptr_c ) ptr_c = ptr_a; | 706 | if( ptr_a > ptr_c ) ptr_c = ptr_a; |
701 | mem = ptr_c - ptr; | 707 | mem = ptr_c - ptr; |
702 | 708 | ||
703 | for( i=0; i<g_connection_count; ++i ) { | 709 | for( i=0; i < g_connection_count; ++i ) { |
704 | if( PROXYPEER_ISCONNECTED(g_connections[i].state) ) { | 710 | if( PROXYPEER_ISCONNECTED(g_connections[i].state) ) { |
705 | void *tmp = malloc( mem ); | 711 | void *tmp = malloc( mem ); |
706 | if( tmp ) { | 712 | if( tmp ) { |
@@ -735,18 +741,30 @@ void livesync_ticker( ) { | |||
735 | } | 741 | } |
736 | 742 | ||
737 | static void livesync_proxytell( uint8_t prefix, uint8_t *info_hash, uint8_t *peer ) { | 743 | static void livesync_proxytell( uint8_t prefix, uint8_t *info_hash, uint8_t *peer ) { |
744 | unsigned int i; | ||
745 | |||
738 | *g_peerbuffer_pos = prefix; | 746 | *g_peerbuffer_pos = prefix; |
739 | memcpy( g_peerbuffer_pos + 1, info_hash, sizeof(ot_hash) - 1 ); | 747 | memcpy( g_peerbuffer_pos + 1, info_hash, sizeof(ot_hash) - 1 ); |
740 | memcpy( g_peerbuffer_pos + sizeof(ot_hash), peer, sizeof(ot_peer) - 1 ); | 748 | memcpy( g_peerbuffer_pos + sizeof(ot_hash), peer, sizeof(ot_peer) - 1 ); |
741 | 749 | ||
742 | g_peerbuffer_pos += sizeof(ot_hash) + sizeof(ot_peer); | 750 | /* Dump info_hash */ |
751 | for( i=0; i<sizeof(ot_hash); ++i ) | ||
752 | printf( "%02X", g_peerbuffer_pos[i] ); | ||
753 | putchar( ':' ); | ||
754 | |||
755 | g_peerbuffer_pos += sizeof(ot_hash); | ||
756 | |||
757 | printf( "%d.%d.%d.%d:%hd (%02X %02X)\n", g_peerbuffer_pos[0], g_peerbuffer_pos[1], g_peerbuffer_pos[2], g_peerbuffer_pos[3], | ||
758 | g_peerbuffer_pos[4] | ( g_peerbuffer_pos[5] << 8 ), g_peerbuffer_pos[6], g_peerbuffer_pos[7] ); | ||
759 | |||
760 | g_peerbuffer_pos += sizeof(ot_peer); | ||
743 | 761 | ||
744 | if( g_peerbuffer_pos >= g_peerbuffer_highwater ) | 762 | if( g_peerbuffer_pos >= g_peerbuffer_highwater ) |
745 | livesync_issue_peersync(); | 763 | livesync_issue_peersync(); |
746 | } | 764 | } |
747 | 765 | ||
748 | static void process_indata( proxy_peer * peer ) { | 766 | static void process_indata( proxy_peer * peer ) { |
749 | int ensuremem, consumed, peers; | 767 | int consumed, peers; |
750 | uint8_t *data = peer->indata, *hash; | 768 | uint8_t *data = peer->indata, *hash; |
751 | uint8_t *dataend = data + peer->indata_length; | 769 | uint8_t *dataend = data + peer->indata_length; |
752 | 770 | ||
@@ -760,21 +778,37 @@ static void process_indata( proxy_peer * peer ) { | |||
760 | peer->packet_tprefix = data[5]; | 778 | peer->packet_tprefix = data[5]; |
761 | peer->packet_tcount = data[6] * 256 + data[7]; | 779 | peer->packet_tcount = data[6] * 256 + data[7]; |
762 | data += 8; | 780 | data += 8; |
781 | printf( "type: %d, prefix: %02X, torrentcount: %d\n", peer->packet_type, peer->packet_tprefix, peer->packet_tcount ); | ||
763 | } | 782 | } |
764 | 783 | ||
765 | /* ensure size for the complete torrent block */ | 784 | next_torrent: |
766 | if( data + 26 > dataend ) break; | 785 | /* Ensure size for the complete torrent block */ |
767 | peers = peer->packet_type ? peer->packet_type : data[19]; | 786 | if( data + sizeof(ot_hash) + OT_IP_SIZE + 3 > dataend ) break; |
768 | ensuremem = 19 + ( peer->packet_type == 0 ) + 7 * peers; | ||
769 | if( data + ensuremem > dataend ) break; | ||
770 | 787 | ||
788 | /* Advance pointer to peer count or peers */ | ||
771 | hash = data; | 789 | hash = data; |
772 | data += 19 + ( peer->packet_type == 0 ); | 790 | data += sizeof(ot_hash) - 1; |
791 | |||
792 | /* Type 0 has peer count encoded before each peers */ | ||
793 | if( peer->packet_type == 0 ) { | ||
794 | int shift = 0; | ||
795 | peers = 0; | ||
796 | do peers |= ( 0x7f & *data ) << ( 7 * shift ); | ||
797 | while ( *(data++) & 0x80 && shift++ < 6 ); | ||
798 | } else | ||
799 | peers = peer->packet_type; | ||
800 | |||
801 | /* Ensure enough data being read to hold all peers */ | ||
802 | if( data + 7 * peers > dataend ) break; | ||
803 | |||
804 | printf( "peers: %d\n", peers ); | ||
773 | 805 | ||
774 | while( peers-- ) { | 806 | while( peers-- ) { |
775 | livesync_proxytell( peer->packet_tprefix, hash, data ); | 807 | livesync_proxytell( peer->packet_tprefix, hash, data ); |
776 | data += 7; | 808 | data += OT_IP_SIZE + 3; |
777 | } | 809 | } |
810 | if( --peer->packet_tcount ) | ||
811 | goto next_torrent; | ||
778 | } | 812 | } |
779 | 813 | ||
780 | consumed = data - peer->indata; | 814 | consumed = data - peer->indata; |