source: branches/encryption/libtransmission/peer-mgr.c @ 3054

Last change on this file since 3054 was 3054, checked in by charles, 15 years ago

use much less memory in the peer manager

  • Property svn:keywords set to Date Rev Author Id
File size: 26.9 KB
Line 
1/*
2 * This file Copyright (C) 2007 Charles Kerr <charles@rebelbase.com>
3 *
4 * This file is licensed by the GPL version 2.  Works owned by the
5 * Transmission project are granted a special exemption to clause 2(b)
6 * so that the bulk of its code can remain under the MIT license.
7 * This exemption does not extend to derived works not owned by
8 * the Transmission project.
9 *
10 * $Id: peer-mgr.c 3054 2007-09-12 23:09:02Z charles $
11 */
12
13#include <assert.h>
14#include <string.h> /* memcpy, memcmp */
15#include <stdlib.h> /* qsort */
16#include <stdio.h> /* printf */
17
18#include "transmission.h"
19#include "handshake.h"
20#include "completion.h"
21#include "net.h"
22#include "peer-io.h"
23#include "peer-mgr.h"
24#include "peer-mgr-private.h"
25#include "peer-msgs.h"
26#include "ptrarray.h"
27#include "trevent.h"
28#include "utils.h"
29
30#define MINUTES_TO_MSEC(N) ((N) * 60 * 1000)
31
32/* how frequently to change which peers are choked */
33#define RECHOKE_PERIOD_SECONDS (MINUTES_TO_MSEC(10))
34
35/* how many downloaders to unchoke per-torrent.
36 * http://wiki.theory.org/BitTorrentSpecification#Choking_and_Optimistic_Unchoking */
37#define NUM_DOWNLOADERS_TO_UNCHOKE 4
38
39/* across all torrents, how many peers maximum do we want connected? */
40#define MAX_CONNECTED_PEERS 80
41
42/**
43***
44**/
45
46struct tr_block
47{
48    unsigned int have          : 1;
49    unsigned int dnd           : 1;
50    unsigned int low_priority  : 1;
51    unsigned int high_priority : 1;
52    uint8_t requestCount;
53    uint8_t scarcity;
54    uint32_t block;
55};
56
57#define MAX_SCARCITY UINT8_MAX
58#define MAX_REQ_COUNT UINT8_MAX
59
60static void
61incrementReqCount( struct tr_block * block )
62{
63    assert( block != NULL );
64
65    if( block->requestCount < MAX_REQ_COUNT )
66        block->requestCount++;
67}
68
69static void
70incrementScarcity( struct tr_block * block )
71{
72    assert( block != NULL );
73
74    if( block->scarcity < MAX_SCARCITY )
75        block->scarcity++;
76}
77
78static int
79compareBlockByIndex( const void * va, const void * vb )
80{
81    const struct tr_block * a = (const struct tr_block *) va;
82    const struct tr_block * b = (const struct tr_block *) vb;
83    return tr_compareUint32( a->block, b->block );
84}
85
86static int
87compareBlockByInterest( const void * va, const void * vb )
88{
89    const struct tr_block * a = (const struct tr_block *) va;
90    const struct tr_block * b = (const struct tr_block *) vb;
91    int i;
92
93    if( a->dnd != b->dnd )
94        return a->dnd ? 1 : -1;
95
96    if( a->have != b->have )
97        return a->have ? 1 : -1;
98
99    if(( i = tr_compareUint8( a->requestCount, b->requestCount )))
100        return i;
101
102    if( a->high_priority != b->high_priority )
103        return a->high_priority ? -1 : 1;
104
105    if( a->low_priority != b->low_priority )
106        return a->low_priority ? 1 : -1;
107
108    if(( i = tr_compareUint16( a->scarcity, b->scarcity )))
109        return i;
110
111    if(( i = tr_compareUint32( a->block, b->block )))
112        return i;
113
114    return 0;
115}
116
117/**
118***
119**/
120
121typedef struct
122{
123    uint8_t hash[SHA_DIGEST_LENGTH];
124    tr_ptrArray * peers; /* tr_peer */
125    tr_timer * chokeTimer;
126    tr_timer * refillTimer;
127    tr_torrent * tor;
128
129    struct tr_block * blocks;
130    uint32_t blockCount;
131
132    unsigned int isRunning : 1;
133
134    struct tr_peerMgr * manager;
135}
136Torrent;
137
138struct tr_peerMgr
139{
140    tr_handle * handle;
141    tr_ptrArray * torrents; /* Torrent */
142    int connectionCount;
143    tr_ptrArray * handshakes; /* in-process */
144};
145
146/**
147***
148**/
149
150static int
151torrentCompare( const void * va, const void * vb )
152{
153    const Torrent * a = (const Torrent*) va;
154    const Torrent * b = (const Torrent*) vb;
155    return memcmp( a->hash, b->hash, SHA_DIGEST_LENGTH );
156}
157
158static int
159torrentCompareToHash( const void * va, const void * vb )
160{
161    const Torrent * a = (const Torrent*) va;
162    const uint8_t * b_hash = (const uint8_t*) vb;
163    return memcmp( a->hash, b_hash, SHA_DIGEST_LENGTH );
164}
165
166static Torrent*
167getExistingTorrent( tr_peerMgr * manager, const uint8_t * hash )
168{
169    return (Torrent*) tr_ptrArrayFindSorted( manager->torrents,
170                                             hash,
171                                             torrentCompareToHash );
172}
173
174static int chokePulse( void * vtorrent );
175
176static int
177peerCompare( const void * va, const void * vb )
178{
179    const tr_peer * a = (const tr_peer *) va;
180    const tr_peer * b = (const tr_peer *) vb;
181    return memcmp( &a->in_addr, &b->in_addr, sizeof(struct in_addr) );
182}
183
184static int
185peerCompareToAddr( const void * va, const void * vb )
186{
187    const tr_peer * a = (const tr_peer *) va;
188    const struct in_addr * b = (const struct in_addr *) vb;
189    return memcmp( &a->in_addr, b, sizeof(struct in_addr) );
190}
191
192static tr_peer*
193getExistingPeer( Torrent * torrent, const struct in_addr * in_addr )
194{
195    return (tr_peer*) tr_ptrArrayFindSorted( torrent->peers,
196                                             in_addr,
197                                             peerCompareToAddr );
198}
199
200static tr_peer*
201getPeer( Torrent * torrent, const struct in_addr * in_addr )
202{
203    tr_peer * peer = getExistingPeer( torrent, in_addr );
204    if( peer == NULL )
205    {
206        peer = tr_new0( tr_peer, 1 );
207        memcpy( &peer->in_addr, in_addr, sizeof(struct in_addr) );
208        tr_ptrArrayInsertSorted( torrent->peers, peer, peerCompare );
209fprintf( stderr, "getPeer: torrent %p now has %d peers\n", torrent, tr_ptrArraySize(torrent->peers) );
210    }
211    return peer;
212}
213
214static void
215disconnectPeer( tr_peer * peer )
216{
217    assert( peer != NULL );
218
219    tr_peerIoFree( peer->io );
220    peer->io = NULL;
221
222    if( peer->msgs != NULL )
223    {
224fprintf( stderr, "PUB unsub peer %p from msgs %p\n", peer, peer->msgs );
225        tr_peerMsgsUnsubscribe( peer->msgs, peer->msgsTag );
226        tr_peerMsgsFree( peer->msgs );
227        peer->msgs = NULL;
228    }
229
230    tr_bitfieldFree( peer->have );
231    peer->have = NULL;
232
233    tr_bitfieldFree( peer->blame );
234    peer->blame = NULL;
235
236    tr_bitfieldFree( peer->banned );
237    peer->banned = NULL;
238}
239
240static void
241freePeer( tr_peer * peer )
242{
243    disconnectPeer( peer );
244    tr_free( peer->client );
245    tr_free( peer );
246}
247
248static void
249freeTorrent( tr_peerMgr * manager, Torrent * t )
250{
251    int i, size;
252    tr_peer ** peers;
253fprintf( stderr, "timer freeTorrent %p\n", t );
254
255    assert( manager != NULL );
256    assert( t != NULL );
257    assert( t->peers != NULL );
258
259    peers = (tr_peer **) tr_ptrArrayPeek( t->peers, &size );
260fprintf( stderr, "timer CHOKE %p being freed\n", t->chokeTimer );
261    tr_timerFree( &t->chokeTimer );
262fprintf( stderr, "timer REFILL %p being freed\n", t->refillTimer );
263    tr_timerFree( &t->refillTimer );
264    for( i=0; i<size; ++i )
265        freePeer( peers[i] );
266    tr_ptrArrayFree( t->peers );
267    tr_ptrArrayRemoveSorted( manager->torrents, t, torrentCompare );
268    tr_free( t->blocks );
269    tr_free( t );
270}
271
272/**
273***
274**/
275
276tr_peerMgr*
277tr_peerMgrNew( tr_handle * handle )
278{
279    tr_peerMgr * m = tr_new0( tr_peerMgr, 1 );
280    m->handle = handle;
281    m->torrents = tr_ptrArrayNew( );
282    m->handshakes = tr_ptrArrayNew( );
283    return m;
284}
285
286void
287tr_peerMgrFree( tr_peerMgr * manager )
288{
289    int i, n;
290fprintf( stderr, "timer peerMgrFree\n" );
291
292    while( !tr_ptrArrayEmpty( manager->torrents ) )
293        freeTorrent( manager, (Torrent*)tr_ptrArrayNth( manager->torrents, 0) );
294    tr_ptrArrayFree( manager->torrents );
295
296    for( i=0, n=tr_ptrArraySize(manager->handshakes); i<n; ++i )
297        tr_handshakeAbort( (tr_handshake*) tr_ptrArrayNth( manager->handshakes, i) );
298    tr_ptrArrayFree( manager->handshakes );
299
300    tr_free( manager );
301}
302
303/**
304***
305**/
306
307static tr_peer**
308getConnectedPeers( Torrent * t, int * setmeCount )
309{
310    int i, peerCount, connectionCount;
311    tr_peer **peers = (tr_peer **) tr_ptrArrayPeek( t->peers, &peerCount );
312    tr_peer **ret = tr_new( tr_peer*, peerCount );
313
314    for( i=connectionCount=0; i<peerCount; ++i )
315        if( peers[i]->msgs != NULL )
316            ret[connectionCount++] = peers[i];
317
318    *setmeCount = connectionCount;
319    return ret;
320}
321
322static void
323populateBlockArray( Torrent * t )
324{
325    uint32_t i;
326    struct tr_block * b;
327    const tr_torrent * tor = t->tor;
328
329    t->blockCount = tor->blockCount;
330    t->blocks = b = tr_new( struct tr_block, t->blockCount );
331
332    for( i=0; i<t->blockCount; ++i, ++b )
333    {
334        const int index = tr_torBlockPiece( tor, i );
335        b->have = tr_cpBlockIsComplete( tor->completion, i ) ? 1 : 0;
336        b->dnd = tor->info.pieces[index].dnd ? 1 : 0;
337        b->low_priority = tor->info.pieces[index].priority == TR_PRI_LOW;
338        b->high_priority = tor->info.pieces[index].priority == TR_PRI_HIGH;
339        b->requestCount = 0;
340        b->scarcity = 0;
341        b->block = i;
342    }
343}
344
345static int
346refillPulse( void * vtorrent )
347{
348    uint32_t i;
349    int size;
350    Torrent * t = (Torrent *) vtorrent;
351    tr_peer ** peers;
352    const int isSeeding = tr_cpGetStatus( t->tor->completion ) != TR_CP_INCOMPLETE;
353    const int wantToRefill = t->isRunning && !isSeeding;
354
355    if( !wantToRefill && t->blocks!=NULL ) /* torrent has stopped or switched to seeding */
356    {
357        tr_free( t->blocks );
358        t->blocks = NULL;
359        t->blockCount = 0;
360    }
361    else if( wantToRefill && t->blocks==NULL ) /* torrent has started or switched to leeching */
362    {
363        populateBlockArray( t );
364    }
365
366    peers = getConnectedPeers( t, &size );
367    if( size>0 )
368    {
369        /* sort the blocks by interest */
370        fprintf( stderr, "sorting [%s] blocks by interest...", t->tor->info.name );
371        qsort( t->blocks, t->blockCount, sizeof(struct tr_block), compareBlockByInterest );
372        fprintf( stderr, "done\n" );
373
374        /* walk through all the most interesting blocks */
375        for( i=0; i<t->blockCount; ++i )
376        {
377            const uint32_t b = t->blocks[i].block;
378            const uint32_t index = tr_torBlockPiece( t->tor, b );
379            const uint32_t begin = ( b * t->tor->blockSize )-( index * t->tor->info.pieceSize );
380            const uint32_t length = tr_torBlockCountBytes( t->tor, (int)b );
381            int j;
382
383            if( t->blocks[i].have || t->blocks[i].dnd )
384                continue;
385
386            if( !size ) /* all peers full */
387                break;
388
389            /* find a peer who can ask for this block */
390            for( j=0; j<size; )
391            {
392                const int val = tr_peerMsgsAddRequest( peers[j]->msgs, index, begin, length );
393                if( val == TR_ADDREQ_FULL ) {
394                    fprintf( stderr, "peer %d of %d is full\n", (int)j, size );
395                    peers[j] = peers[--size];
396                }
397                else if( val == TR_ADDREQ_MISSING ) {
398                    fprintf( stderr, "peer doesn't have it\n" );
399                    ++j;
400                }
401                else if( val == TR_ADDREQ_OK ) {
402                    fprintf( stderr, "peer %d took the request for block %d\n", j, i );
403                    incrementReqCount( &t->blocks[i] );
404                    j = size;
405                }
406            }
407        }
408
409        /* put the blocks back the way we found them */
410        qsort( t->blocks, t->blockCount, sizeof(struct tr_block), compareBlockByIndex );
411    }
412
413    /* cleanup */
414    tr_free( peers );
415
416    /* let the timer expire */
417    t->refillTimer = NULL;
418    return FALSE;
419}
420
421static void
422ensureRefillTag( Torrent * t )
423{
424    if( t->refillTimer == NULL ) {
425        t->refillTimer = tr_timerNew( t->manager->handle, refillPulse, t, 750 );
426        fprintf( stderr, "timer REFILL %p is new\n", t->refillTimer );
427    }
428}
429
430static void
431msgsCallbackFunc( void * source UNUSED, void * vevent, void * vt )
432{
433    Torrent * t = (Torrent *) vt;
434    const tr_peermsgs_event * e = (const tr_peermsgs_event *) vevent;
435
436    switch( e->eventType )
437    {
438        case TR_PEERMSG_GOT_BITFIELD: {
439            const uint32_t begin = 0;
440            const uint32_t end = begin + t->tor->info.pieceCount;
441            uint32_t i;
442            for( i=begin; t->blocks!=NULL && i<end; ++i ) {
443                if( !tr_bitfieldHas( e->bitfield, i ) )
444                    continue;
445                assert( t->blocks[i].block == i );
446                incrementScarcity( &t->blocks[i] );
447            }
448            break;
449        }
450
451        case TR_PEERMSG_GOT_HAVE: {
452            const uint32_t begin = tr_torPieceFirstBlock( t->tor, e->pieceIndex );
453            const uint32_t end = begin + tr_torPieceCountBlocks( t->tor, (int)e->pieceIndex );
454            uint32_t i;
455            for( i=begin; t->blocks!=NULL && i<end; ++i ) {
456                assert( t->blocks[i].block == i );
457                incrementScarcity( &t->blocks[i] );
458            }
459            break;
460        }
461
462        case TR_PEERMSG_GOT_BLOCK: {
463            uint32_t i = e->blockIndex;
464            if( t->blocks != NULL ) {
465                assert( t->blocks[i].block == i );
466                t->blocks[i].have = 1;
467            }
468            break;
469        }
470
471        case TR_PEERMSG_GOT_PEX:
472            /* FIXME */
473            break;
474
475        case TR_PEERMSG_GOT_ERROR:
476            /* FIXME */
477            break;
478
479        case TR_PEERMSG_BLOCKS_RUNNING_LOW:
480            ensureRefillTag( t );
481            break;
482
483        default:
484            assert(0);
485    }
486}
487
488static void
489myHandshakeDoneCB( tr_handshake * handshake, tr_peerIo * io, int isConnected, void * vmanager )
490{
491    int ok = isConnected;
492    uint16_t port;
493    const struct in_addr * in_addr;
494    tr_peerMgr * manager = (tr_peerMgr*) vmanager;
495    const uint8_t * hash = NULL;
496    Torrent * t;
497
498    assert( io != NULL );
499
500    tr_ptrArrayRemoveSorted( manager->handshakes, handshake, tr_comparePointers );
501
502    in_addr = tr_peerIoGetAddress( io, &port );
503
504    if( !tr_peerIoHasTorrentHash( io ) ) /* incoming connection gone wrong? */
505    {
506        tr_peerIoFree( io );
507        --manager->connectionCount;
508        return;
509    }
510
511    hash = tr_peerIoGetTorrentHash( io );
512    t = getExistingTorrent( manager, hash );
513    if( !t )
514    {
515        tr_peerIoFree( io );
516        --manager->connectionCount;
517        return;
518    }
519
520    fprintf( stderr, "peer-mgr: torrent [%s] finished a handshake; isConnected is %d\n", t->tor->info.name, isConnected );
521
522    /* if we couldn't connect or were snubbed,
523     * the peer's probably not worth remembering. */
524    if( !ok ) {
525        tr_peer * peer = getExistingPeer( t, in_addr );
526        fprintf( stderr, "peer-mgr: torrent [%s] got a bad one, and you know what? fuck them.\n", t->tor->info.name );
527        if( peer ) {
528            tr_ptrArrayRemoveSorted( t->peers, peer, peerCompare );
529            freePeer( peer );
530        } else  {
531            tr_peerIoFree( io );
532        }
533        --manager->connectionCount;
534        return;
535    }
536
537#if 0
538    /* ONLY DO THIS TEST FOR INCOMING CONNECTIONS */
539    /* check for duplicates */
540    if( getExistingPeer( t, in_addr ) ) {
541        tr_dbg( "dropping a duplicate connection... dropping." );
542        tr_peerIoFree( io );
543        return;
544    }
545#endif
546
547    if( 1 ) {
548        tr_peer * peer = getPeer( t, in_addr );
549        peer->port = port;
550        peer->io = io;
551        peer->msgs = tr_peerMsgsNew( t->tor, peer );
552fprintf( stderr, "PUB sub peer %p to msgs %p\n", peer, peer->msgs );
553        peer->msgsTag = tr_peerMsgsSubscribe( peer->msgs, msgsCallbackFunc, t );
554        chokePulse( t );
555    }
556}
557
558void
559tr_peerMgrAddIncoming( tr_peerMgr      * manager,
560                       struct in_addr  * addr,
561                       int               socket )
562{
563    tr_handshake * handshake;
564
565    ++manager->connectionCount;
566
567    handshake = tr_handshakeNew( tr_peerIoNewIncoming( manager->handle, addr, socket ),
568                                 HANDSHAKE_ENCRYPTION_PREFERRED,
569                                 myHandshakeDoneCB,
570                                 manager );
571
572    tr_ptrArrayInsertSorted( manager->handshakes, handshake, tr_comparePointers );
573}
574
575static void
576maybeConnect( tr_peerMgr * manager, Torrent * t, tr_peer * peer )
577{
578    if( tr_peerMgrIsAcceptingConnections( manager ) )
579    {
580        fprintf( stderr, "peer-mgr: torrent [%s] is handshaking with a new peer %08x:%04x\n",
581                 t->tor->info.name,
582                 (uint32_t) peer->in_addr.s_addr, peer->port );
583
584        ++manager->connectionCount;
585
586        peer->io = tr_peerIoNewOutgoing( manager->handle,
587                                         &peer->in_addr,
588                                         peer->port,
589                                         t->hash );
590
591        tr_handshakeNew( peer->io, HANDSHAKE_ENCRYPTION_PREFERRED,
592                         myHandshakeDoneCB, manager );
593    }
594}
595
596void
597tr_peerMgrAddPex( tr_peerMgr     * manager,
598                  const uint8_t  * torrentHash,
599                  int              from,
600                  const tr_pex   * pex,
601                  int              pexCount )
602{
603    int i;
604    const tr_pex * walk = pex;
605    Torrent * t = getExistingTorrent( manager, torrentHash );
606    for( i=0; i<pexCount; ++i )
607    {
608        tr_peer * peer = getPeer( t, &walk->in_addr );
609        peer->port = walk->port;
610        peer->from = from;
611        maybeConnect( manager, t, peer );
612    }
613}
614
615void
616tr_peerMgrAddPeers( tr_peerMgr    * manager,
617                    const uint8_t * torrentHash,
618                    int             from,
619                    const uint8_t * peerCompact,
620                    int             peerCount )
621{
622    int i;
623    const uint8_t * walk = peerCompact;
624    Torrent * t = getExistingTorrent( manager, torrentHash );
625    for( i=0; i<peerCount; ++i )
626    {
627        tr_peer * peer;
628        struct in_addr addr;
629        uint16_t port;
630        memcpy( &addr, walk, 4 ); walk += 4;
631        memcpy( &port, walk, 2 ); walk += 2;
632        peer = getPeer( t, &addr );
633        peer->port = port;
634        peer->from = from;
635        maybeConnect( manager, t, peer );
636    }
637}
638
639/**
640***
641**/
642
643int
644tr_peerMgrIsAcceptingConnections( const tr_peerMgr * manager )
645{
646    return manager->connectionCount < MAX_CONNECTED_PEERS;
647}
648
649void
650tr_peerMgrSetBlame( tr_peerMgr     * manager UNUSED,
651                    const uint8_t  * torrentHash UNUSED,
652                    int              pieceIndex UNUSED,
653                    int              success UNUSED )
654{
655    assert( 0 );
656}
657
658int
659tr_pexCompare( const void * va, const void * vb )
660{
661    const tr_pex * a = (const tr_pex *) va;
662    const tr_pex * b = (const tr_pex *) vb;
663    int i = memcmp( &a->in_addr, &b->in_addr, sizeof(struct in_addr) );
664    if( i ) return i;
665    if( a->port < b->port ) return -1;
666    if( a->port > b->port ) return 1;
667    return 0;
668}
669
670int tr_pexCompare( const void * a, const void * b );
671
672
673int
674tr_peerMgrGetPeers( tr_peerMgr      * manager,
675                    const uint8_t   * torrentHash,
676                    tr_pex         ** setme_pex )
677{
678    const Torrent * t = getExistingTorrent( (tr_peerMgr*)manager, torrentHash );
679    int i, peerCount;
680    const tr_peer ** peers = (const tr_peer **) tr_ptrArrayPeek( t->peers, &peerCount );
681    tr_pex * pex = tr_new( tr_pex, peerCount );
682    tr_pex * walk = pex;
683
684    for( i=0; i<peerCount; ++i, ++walk )
685    {
686        walk->in_addr = peers[i]->in_addr;
687        walk->port = peers[i]->port;
688        walk->flags = '\0'; /* FIXME */
689    }
690
691    assert( ( walk - pex ) == peerCount );
692    qsort( pex, peerCount, sizeof(tr_pex), tr_pexCompare );
693    *setme_pex = pex;
694    return peerCount;
695}
696
697void
698tr_peerMgrStartTorrent( tr_peerMgr     * manager UNUSED,
699                        const uint8_t  * torrentHash UNUSED)
700{
701    Torrent * t = getExistingTorrent( manager, torrentHash );
702    t->isRunning = TRUE;
703}
704
705void
706tr_peerMgrStopTorrent( tr_peerMgr     * manager,
707                       const uint8_t  * torrentHash)
708{
709    int i, size;
710    Torrent * t = getExistingTorrent( manager, torrentHash );
711    tr_peer ** peers = getConnectedPeers( t, &size );
712
713    t->isRunning = FALSE;
714
715    for( i=0; i<size; ++i )
716        disconnectPeer( peers[i] );
717}
718
719void
720tr_peerMgrUpdateCompletion( tr_peerMgr     * manager,
721                            const uint8_t  * torrentHash )
722{
723    uint32_t i;
724    Torrent * t = getExistingTorrent( manager, torrentHash );
725
726    for( i=0; t->blocks!=NULL && i<t->blockCount; ++i ) {
727        assert( t->blocks[i].block == i );
728        t->blocks[i].have = tr_cpBlockIsComplete( t->tor->completion, i ) ? 1 : 0;
729    }
730}
731
732void
733tr_peerMgrAddTorrent( tr_peerMgr * manager,
734                      tr_torrent * tor )
735{
736    Torrent * t;
737
738    assert( tor != NULL );
739    assert( getExistingTorrent( manager, tor->info.hash ) == NULL );
740
741    t = tr_new0( Torrent, 1 );
742    t->manager = manager;
743    t->tor = tor;
744    t->peers = tr_ptrArrayNew( );
745    t->chokeTimer = tr_timerNew( manager->handle, chokePulse, t, RECHOKE_PERIOD_SECONDS );
746fprintf( stderr, "timer CHOKE %p is new\n", t->chokeTimer );
747
748    memcpy( t->hash, tor->info.hash, SHA_DIGEST_LENGTH );
749    tr_ptrArrayInsertSorted( manager->torrents, t, torrentCompare );
750}
751
752void
753tr_peerMgrRemoveTorrent( tr_peerMgr     * manager,
754                         const uint8_t  * torrentHash )
755{
756    Torrent * t = getExistingTorrent( manager, torrentHash );
757    assert( t != NULL );
758    tr_peerMgrStopTorrent( manager, torrentHash );
759    freeTorrent( manager, t );
760}
761
762void
763tr_peerMgrTorrentAvailability( const tr_peerMgr * manager,
764                               const uint8_t    * torrentHash,
765                               int8_t           * tab,
766                               int                tabCount )
767{
768    int i;
769    const Torrent * t = getExistingTorrent( (tr_peerMgr*)manager, torrentHash );
770    const tr_torrent * tor = t->tor;
771    const float interval = tor->info.pieceCount / (float)tabCount;
772
773    for( i=0; i<tabCount; ++i )
774    {
775        const int piece = i * interval;
776
777        if( tor == NULL )
778            tab[i] = 0;
779        else if( tr_cpPieceIsComplete( tor->completion, piece ) )
780            tab[i] = -1;
781        else {
782            int j, peerCount;
783            const tr_peer ** peers = (const tr_peer **) tr_ptrArrayPeek( t->peers, &peerCount );
784            for( j=0; j<peerCount; ++j )
785                if( tr_bitfieldHas( peers[j]->have, i ) )
786                    ++tab[i];
787        }
788    }
789}
790
791
792void
793tr_peerMgrTorrentStats( const tr_peerMgr * manager,
794                        const uint8_t    * torrentHash,
795                        int              * setmePeersTotal,
796                        int              * setmePeersConnected,
797                        int              * setmePeersSendingToUs,
798                        int              * setmePeersGettingFromUs,
799                        int              * setmePeersFrom )
800{
801    int i, size;
802    const Torrent * t = getExistingTorrent( (tr_peerMgr*)manager, torrentHash );
803    const tr_peer ** peers = (const tr_peer **) tr_ptrArrayPeek( t->peers, &size );
804
805    *setmePeersTotal          = size;
806    *setmePeersConnected      = 0;
807    *setmePeersSendingToUs    = 0;
808    *setmePeersGettingFromUs  = 0;
809
810    for( i=0; i<size; ++i )
811    {
812        const tr_peer * peer = peers[i];
813
814        if( peer->io == NULL ) /* not connected */
815            continue;
816
817        ++*setmePeersConnected;
818
819        ++setmePeersFrom[peer->from];
820
821        if( tr_peerIoGetRateToPeer( peer->io ) > 0.01 )
822            ++*setmePeersGettingFromUs;
823
824        if( tr_peerIoGetRateToClient( peer->io ) > 0.01 )
825            ++*setmePeersSendingToUs;
826    }
827}
828
829struct tr_peer_stat *
830tr_peerMgrPeerStats( const tr_peerMgr  * manager,
831                     const uint8_t     * torrentHash,
832                     int               * setmeCount UNUSED )
833{
834    int i, size;
835    const Torrent * t = getExistingTorrent( (tr_peerMgr*)manager, torrentHash );
836    const tr_peer ** peers = (const tr_peer **) tr_ptrArrayPeek( t->peers, &size );
837    tr_peer_stat * ret;
838
839    ret = tr_new0( tr_peer_stat, size );
840
841    for( i=0; i<size; ++i )
842    {
843        const tr_peer * peer = peers[i];
844        const int live = peer->io != NULL;
845        tr_peer_stat * stat = ret + i;
846
847        tr_netNtop( &peer->in_addr, stat->addr, sizeof(stat->addr) );
848        stat->port             = peer->port;
849        stat->from             = peer->from;
850        stat->client           = peer->client;
851        stat->progress         = peer->progress;
852        stat->isConnected      = live ? 1 : 0;
853        stat->isEncrypted      = tr_peerIoIsEncrypted( peer->io ) ? 1 : 0;
854        stat->uploadToRate     = tr_peerIoGetRateToPeer( peer->io );
855        stat->downloadFromRate = tr_peerIoGetRateToClient( peer->io );
856        stat->isDownloading    = stat->uploadToRate > 0.01;
857        stat->isUploading      = stat->downloadFromRate > 0.01;
858    }
859
860    *setmeCount = size;
861    return ret;
862}
863
864void
865tr_peerMgrDisablePex( tr_peerMgr    * manager,
866                      const uint8_t * torrentHash,
867                      int             disable)
868{
869    Torrent * t = getExistingTorrent( manager, torrentHash );
870    tr_torrent * tor = t->tor;
871
872    if( ( tor->pexDisabled != disable ) && ! ( TR_FLAG_PRIVATE & tor->info.flags ) )
873    {
874        int i, size;
875        tr_peer ** peers = (tr_peer **) tr_ptrArrayPeek( t->peers, &size );
876        for( i=0; i<size; ++i ) {
877            peers[i]->pexEnabled = disable ? 0 : 1;
878            peers[i]->lastPexTime = 0;
879        }
880
881        tor->pexDisabled = disable;
882    }
883}
884
885/**
886***
887**/
888
889typedef struct
890{
891    tr_peer * peer;
892    float rate;
893    int isInterested;
894}
895ChokeData;
896
897static int
898compareChokeByRate( const void * va, const void * vb )
899{
900    const ChokeData * a = ( const ChokeData * ) va;
901    const ChokeData * b = ( const ChokeData * ) vb;
902    if( a->rate > b->rate ) return -1;
903    if( a->rate < b->rate ) return 1;
904    return 0;
905}
906
907static int
908compareChokeByDownloader( const void * va, const void * vb )
909{
910    const ChokeData * a = ( const ChokeData * ) va;
911    const ChokeData * b = ( const ChokeData * ) vb;
912
913    /* primary key: interest */
914    if(  a->isInterested && !b->isInterested ) return -1;
915    if( !a->isInterested &&  b->isInterested ) return 1;
916
917    /* second key: rate */
918    return compareChokeByRate( va, vb );
919}
920
921static int
922chokePulse( void * vtorrent )
923{
924    Torrent * t = (Torrent *) vtorrent;
925    int i, size, unchoked;
926    const int done = tr_cpGetStatus( t->tor->completion ) != TR_CP_INCOMPLETE;
927    float bestDownloaderRate;
928    ChokeData * data;
929    tr_peer ** peers = getConnectedPeers( t, &size );
930
931fprintf( stderr, "rechoking torrent %p, with %d peers\n", t, size );
932
933    if( size < 1 )
934        return TRUE;
935
936    data = tr_new( ChokeData, size );
937    for( i=0; i<size; ++i ) {
938        data[i].peer = peers[i];
939        data[i].isInterested = peers[i]->peerIsInterested;
940        data[i].rate = done ? tr_peerIoGetRateToPeer( peers[i]->io )
941                            : tr_peerIoGetRateToClient( peers[i]->io );
942    }
943
944    /* find the best downloaders and unchoke them */
945    qsort( data, size, sizeof(ChokeData), compareChokeByDownloader );
946    bestDownloaderRate = data[0].rate;
947    for( i=unchoked=0; i<size && unchoked<NUM_DOWNLOADERS_TO_UNCHOKE; ++i ) {
948        tr_peerMsgsSetChoke( data[i].peer->msgs, FALSE );
949        ++unchoked;
950    }
951    memmove( data, data+i, sizeof(ChokeData)*(size-i) );
952    size -= i;
953
954    /* of those remaining, unchoke those that are faster than the downloaders */
955    qsort( data, size, sizeof(ChokeData), compareChokeByRate );
956    for( i=0; i<size && data[i].rate >= bestDownloaderRate; ++i )
957        tr_peerMsgsSetChoke( data[i].peer->msgs, FALSE );
958    memmove( data, data+i, sizeof(ChokeData)*(size-i) );
959    size -= i;
960
961    /* of those remaining, optimistically unchoke one; choke the rest */
962    if( size > 0 ) {
963        const int optimistic = tr_rand( size );
964        for( i=0; i<size; ++i )
965            tr_peerMsgsSetChoke( data[i].peer->msgs, i!=optimistic );
966    }
967
968    /* cleanup */
969    tr_free( data );
970    tr_free( peers );
971    return TRUE;
972}
Note: See TracBrowser for help on using the repository browser.