source: trunk/libtransmission/peer-msgs.c @ 11945

Last change on this file since 11945 was 11945, checked in by jordan, 11 years ago

if we successfully finish a handshake using uTP, mark the peer as supporting uTP

  • Property svn:keywords set to Date Rev Author Id
File size: 71.4 KB
Line 
1/*
2 * This file Copyright (C) Mnemosyne LLC
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-msgs.c 11945 2011-02-18 00:41:06Z jordan $
11 */
12
13#include <assert.h>
14#include <errno.h>
15#include <limits.h> /* INT_MAX */
16#include <stdarg.h>
17#include <stdio.h>
18#include <stdlib.h>
19#include <string.h>
20
21#include <event2/bufferevent.h>
22#include <event2/event.h>
23
24#include "transmission.h"
25#include "bencode.h"
26#include "cache.h"
27#include "completion.h"
28#include "crypto.h"
29#ifdef WIN32
30#include "net.h" /* for ECONN */
31#endif
32#include "peer-io.h"
33#include "peer-mgr.h"
34#include "peer-msgs.h"
35#include "session.h"
36#include "stats.h"
37#include "torrent.h"
38#include "torrent-magnet.h"
39#include "tr-dht.h"
40#include "utils.h"
41#include "version.h"
42
43/**
44***
45**/
46
47enum
48{
49    BT_CHOKE                = 0,
50    BT_UNCHOKE              = 1,
51    BT_INTERESTED           = 2,
52    BT_NOT_INTERESTED       = 3,
53    BT_HAVE                 = 4,
54    BT_BITFIELD             = 5,
55    BT_REQUEST              = 6,
56    BT_PIECE                = 7,
57    BT_CANCEL               = 8,
58    BT_PORT                 = 9,
59
60    BT_FEXT_SUGGEST         = 13,
61    BT_FEXT_HAVE_ALL        = 14,
62    BT_FEXT_HAVE_NONE       = 15,
63    BT_FEXT_REJECT          = 16,
64    BT_FEXT_ALLOWED_FAST    = 17,
65
66    BT_LTEP                 = 20,
67
68    LTEP_HANDSHAKE          = 0,
69
70    UT_PEX_ID               = 1,
71    UT_METADATA_ID          = 3,
72
73    MAX_PEX_PEER_COUNT      = 50,
74
75    MIN_CHOKE_PERIOD_SEC    = 10,
76
77    /* idle seconds before we send a keepalive */
78    KEEPALIVE_INTERVAL_SECS = 100,
79
80    PEX_INTERVAL_SECS       = 90, /* sec between sendPex() calls */
81
82    REQQ                    = 512,
83
84    METADATA_REQQ           = 64,
85
86    /* used in lowering the outMessages queue period */
87    IMMEDIATE_PRIORITY_INTERVAL_SECS = 0,
88    HIGH_PRIORITY_INTERVAL_SECS = 2,
89    LOW_PRIORITY_INTERVAL_SECS = 10,
90
91    /* number of pieces to remove from the bitfield when
92     * lazy bitfields are turned on */
93    LAZY_PIECE_COUNT = 26,
94
95    /* number of pieces we'll allow in our fast set */
96    MAX_FAST_SET_SIZE = 3,
97
98    /* defined in BEP #9 */
99    METADATA_MSG_TYPE_REQUEST = 0,
100    METADATA_MSG_TYPE_DATA = 1,
101    METADATA_MSG_TYPE_REJECT = 2
102};
103
104enum
105{
106    AWAITING_BT_LENGTH,
107    AWAITING_BT_ID,
108    AWAITING_BT_MESSAGE,
109    AWAITING_BT_PIECE
110};
111
112/**
113***
114**/
115
116struct peer_request
117{
118    uint32_t    index;
119    uint32_t    offset;
120    uint32_t    length;
121};
122
123static uint32_t
124getBlockOffsetInPiece( const tr_torrent * tor, uint64_t b )
125{
126    const uint64_t piecePos = tor->info.pieceSize * tr_torBlockPiece( tor, b );
127    const uint64_t blockPos = tor->blockSize * b;
128    assert( blockPos >= piecePos );
129    return (uint32_t)( blockPos - piecePos );
130}
131
132static void
133blockToReq( const tr_torrent     * tor,
134            tr_block_index_t       block,
135            struct peer_request  * setme )
136{
137    assert( setme != NULL );
138
139    setme->index = tr_torBlockPiece( tor, block );
140    setme->offset = getBlockOffsetInPiece( tor, block );
141    setme->length = tr_torBlockCountBytes( tor, block );
142}
143
144/**
145***
146**/
147
148/* this is raw, unchanged data from the peer regarding
149 * the current message that it's sending us. */
150struct tr_incoming
151{
152    uint8_t                id;
153    uint32_t               length; /* includes the +1 for id length */
154    struct peer_request    blockReq; /* metadata for incoming blocks */
155    struct evbuffer *      block; /* piece data for incoming blocks */
156};
157
158/**
159 * Low-level communication state information about a connected peer.
160 *
161 * This structure remembers the low-level protocol states that we're
162 * in with this peer, such as active requests, pex messages, and so on.
163 * Its fields are all private to peer-msgs.c.
164 *
165 * Data not directly involved with sending & receiving messages is
166 * stored in tr_peer, where it can be accessed by both peermsgs and
167 * the peer manager.
168 *
169 * @see struct peer_atom
170 * @see tr_peer
171 */
172struct tr_peermsgs
173{
174    tr_bool         peerSupportsPex;
175    tr_bool         peerSupportsMetadataXfer;
176    tr_bool         clientSentLtepHandshake;
177    tr_bool         peerSentLtepHandshake;
178
179    /*tr_bool         haveFastSet;*/
180
181    int             desiredRequestCount;
182
183    int             prefetchCount;
184
185    /* how long the outMessages batch should be allowed to grow before
186     * it's flushed -- some messages (like requests >:) should be sent
187     * very quickly; others aren't as urgent. */
188    int8_t          outMessagesBatchPeriod;
189
190    uint8_t         state;
191    uint8_t         ut_pex_id;
192    uint8_t         ut_metadata_id;
193    uint16_t        pexCount;
194    uint16_t        pexCount6;
195
196#if 0
197    size_t                 fastsetSize;
198    tr_piece_index_t       fastset[MAX_FAST_SET_SIZE];
199#endif
200
201    tr_peer *              peer;
202
203    tr_torrent *           torrent;
204
205    tr_peer_callback      * callback;
206    void                  * callbackData;
207
208    struct evbuffer *      outMessages; /* all the non-piece messages */
209
210    struct peer_request    peerAskedFor[REQQ];
211
212    int                    peerAskedForMetadata[METADATA_REQQ];
213    int                    peerAskedForMetadataCount;
214
215    tr_pex               * pex;
216    tr_pex               * pex6;
217
218    /*time_t                 clientSentPexAt;*/
219    time_t                 clientSentAnythingAt;
220
221    /* when we started batching the outMessages */
222    time_t                outMessagesBatchedAt;
223
224    struct tr_incoming    incoming;
225
226    /* if the peer supports the Extension Protocol in BEP 10 and
227       supplied a reqq argument, it's stored here. Otherwise, the
228       value is zero and should be ignored. */
229    int64_t               reqq;
230
231    struct event        * pexTimer;
232};
233
234/**
235***
236**/
237
238#if 0
239static tr_bitfield*
240getHave( const struct tr_peermsgs * msgs )
241{
242    if( msgs->peer->have == NULL )
243        msgs->peer->have = tr_bitfieldNew( msgs->torrent->info.pieceCount );
244    return msgs->peer->have;
245}
246#endif
247
248static inline tr_session*
249getSession( struct tr_peermsgs * msgs )
250{
251    return msgs->torrent->session;
252}
253
254/**
255***
256**/
257
258static void
259myDebug( const char * file, int line,
260         const struct tr_peermsgs * msgs,
261         const char * fmt, ... )
262{
263    FILE * fp = tr_getLog( );
264
265    if( fp )
266    {
267        va_list           args;
268        char              timestr[64];
269        struct evbuffer * buf = evbuffer_new( );
270        char *            base = tr_basename( file );
271
272        evbuffer_add_printf( buf, "[%s] %s - %s [%s]: ",
273                             tr_getLogTimeStr( timestr, sizeof( timestr ) ),
274                             tr_torrentName( msgs->torrent ),
275                             tr_peerIoGetAddrStr( msgs->peer->io ),
276                             msgs->peer->client );
277        va_start( args, fmt );
278        evbuffer_add_vprintf( buf, fmt, args );
279        va_end( args );
280        evbuffer_add_printf( buf, " (%s:%d)\n", base, line );
281        fputs( (const char*)evbuffer_pullup( buf, -1 ), fp );
282
283        tr_free( base );
284        evbuffer_free( buf );
285    }
286}
287
288#define dbgmsg( msgs, ... ) \
289    do { \
290        if( tr_deepLoggingIsActive( ) ) \
291            myDebug( __FILE__, __LINE__, msgs, __VA_ARGS__ ); \
292    } while( 0 )
293
294/**
295***
296**/
297
298static void
299pokeBatchPeriod( tr_peermsgs * msgs,
300                 int           interval )
301{
302    if( msgs->outMessagesBatchPeriod > interval )
303    {
304        msgs->outMessagesBatchPeriod = interval;
305        dbgmsg( msgs, "lowering batch interval to %d seconds", interval );
306    }
307}
308
309static void
310dbgOutMessageLen( tr_peermsgs * msgs )
311{
312    dbgmsg( msgs, "outMessage size is now %zu", evbuffer_get_length( msgs->outMessages ) );
313}
314
315static void
316protocolSendReject( tr_peermsgs * msgs, const struct peer_request * req )
317{
318    struct evbuffer * out = msgs->outMessages;
319
320    assert( tr_peerIoSupportsFEXT( msgs->peer->io ) );
321
322    evbuffer_add_uint32( out, sizeof( uint8_t ) + 3 * sizeof( uint32_t ) );
323    evbuffer_add_uint8 ( out, BT_FEXT_REJECT );
324    evbuffer_add_uint32( out, req->index );
325    evbuffer_add_uint32( out, req->offset );
326    evbuffer_add_uint32( out, req->length );
327
328    dbgmsg( msgs, "rejecting %u:%u->%u...", req->index, req->offset, req->length );
329    dbgOutMessageLen( msgs );
330}
331
332static void
333protocolSendRequest( tr_peermsgs * msgs, const struct peer_request * req )
334{
335    struct evbuffer * out = msgs->outMessages;
336
337    evbuffer_add_uint32( out, sizeof( uint8_t ) + 3 * sizeof( uint32_t ) );
338    evbuffer_add_uint8 ( out, BT_REQUEST );
339    evbuffer_add_uint32( out, req->index );
340    evbuffer_add_uint32( out, req->offset );
341    evbuffer_add_uint32( out, req->length );
342
343    dbgmsg( msgs, "requesting %u:%u->%u...", req->index, req->offset, req->length );
344    dbgOutMessageLen( msgs );
345    pokeBatchPeriod( msgs, IMMEDIATE_PRIORITY_INTERVAL_SECS );
346}
347
348static void
349protocolSendCancel( tr_peermsgs * msgs, const struct peer_request * req )
350{
351    struct evbuffer * out = msgs->outMessages;
352
353    evbuffer_add_uint32( out, sizeof( uint8_t ) + 3 * sizeof( uint32_t ) );
354    evbuffer_add_uint8 ( out, BT_CANCEL );
355    evbuffer_add_uint32( out, req->index );
356    evbuffer_add_uint32( out, req->offset );
357    evbuffer_add_uint32( out, req->length );
358
359    dbgmsg( msgs, "cancelling %u:%u->%u...", req->index, req->offset, req->length );
360    dbgOutMessageLen( msgs );
361    pokeBatchPeriod( msgs, IMMEDIATE_PRIORITY_INTERVAL_SECS );
362}
363
364static void
365protocolSendPort(tr_peermsgs *msgs, uint16_t port)
366{
367    struct evbuffer * out = msgs->outMessages;
368
369    dbgmsg( msgs, "sending Port %u", port);
370    evbuffer_add_uint32( out, 3 );
371    evbuffer_add_uint8 ( out, BT_PORT );
372    evbuffer_add_uint16( out, port);
373}
374
375static void
376protocolSendHave( tr_peermsgs * msgs, uint32_t index )
377{
378    struct evbuffer * out = msgs->outMessages;
379
380    evbuffer_add_uint32( out, sizeof(uint8_t) + sizeof(uint32_t) );
381    evbuffer_add_uint8 ( out, BT_HAVE );
382    evbuffer_add_uint32( out, index );
383
384    dbgmsg( msgs, "sending Have %u", index );
385    dbgOutMessageLen( msgs );
386    pokeBatchPeriod( msgs, LOW_PRIORITY_INTERVAL_SECS );
387}
388
389#if 0
390static void
391protocolSendAllowedFast( tr_peermsgs * msgs, uint32_t pieceIndex )
392{
393    tr_peerIo       * io  = msgs->peer->io;
394    struct evbuffer * out = msgs->outMessages;
395
396    assert( tr_peerIoSupportsFEXT( msgs->peer->io ) );
397
398    evbuffer_add_uint32( io, out, sizeof(uint8_t) + sizeof(uint32_t) );
399    evbuffer_add_uint8 ( io, out, BT_FEXT_ALLOWED_FAST );
400    evbuffer_add_uint32( io, out, pieceIndex );
401
402    dbgmsg( msgs, "sending Allowed Fast %u...", pieceIndex );
403    dbgOutMessageLen( msgs );
404}
405#endif
406
407static void
408protocolSendChoke( tr_peermsgs * msgs, int choke )
409{
410    struct evbuffer * out = msgs->outMessages;
411
412    evbuffer_add_uint32( out, sizeof( uint8_t ) );
413    evbuffer_add_uint8 ( out, choke ? BT_CHOKE : BT_UNCHOKE );
414
415    dbgmsg( msgs, "sending %s...", choke ? "Choke" : "Unchoke" );
416    dbgOutMessageLen( msgs );
417    pokeBatchPeriod( msgs, IMMEDIATE_PRIORITY_INTERVAL_SECS );
418}
419
420static void
421protocolSendHaveAll( tr_peermsgs * msgs )
422{
423    struct evbuffer * out = msgs->outMessages;
424
425    assert( tr_peerIoSupportsFEXT( msgs->peer->io ) );
426
427    evbuffer_add_uint32( out, sizeof( uint8_t ) );
428    evbuffer_add_uint8 ( out, BT_FEXT_HAVE_ALL );
429
430    dbgmsg( msgs, "sending HAVE_ALL..." );
431    dbgOutMessageLen( msgs );
432    pokeBatchPeriod( msgs, IMMEDIATE_PRIORITY_INTERVAL_SECS );
433}
434
435static void
436protocolSendHaveNone( tr_peermsgs * msgs )
437{
438    struct evbuffer * out = msgs->outMessages;
439
440    assert( tr_peerIoSupportsFEXT( msgs->peer->io ) );
441
442    evbuffer_add_uint32( out, sizeof( uint8_t ) );
443    evbuffer_add_uint8 ( out, BT_FEXT_HAVE_NONE );
444
445    dbgmsg( msgs, "sending HAVE_NONE..." );
446    dbgOutMessageLen( msgs );
447    pokeBatchPeriod( msgs, IMMEDIATE_PRIORITY_INTERVAL_SECS );
448}
449
450/**
451***  EVENTS
452**/
453
454static void
455publish( tr_peermsgs * msgs, tr_peer_event * e )
456{
457    assert( msgs->peer );
458    assert( msgs->peer->msgs == msgs );
459
460    if( msgs->callback != NULL )
461        msgs->callback( msgs->peer, e, msgs->callbackData );
462}
463
464static void
465fireError( tr_peermsgs * msgs, int err )
466{
467    tr_peer_event e = TR_PEER_EVENT_INIT;
468    e.eventType = TR_PEER_ERROR;
469    e.err = err;
470    publish( msgs, &e );
471}
472
473static void
474firePeerProgress( tr_peermsgs * msgs )
475{
476    tr_peer_event e = TR_PEER_EVENT_INIT;
477    e.eventType = TR_PEER_PEER_PROGRESS;
478    e.progress = msgs->peer->progress;
479    publish( msgs, &e );
480}
481
482static void
483fireGotBlock( tr_peermsgs * msgs, const struct peer_request * req )
484{
485    tr_peer_event e = TR_PEER_EVENT_INIT;
486    e.eventType = TR_PEER_CLIENT_GOT_BLOCK;
487    e.pieceIndex = req->index;
488    e.offset = req->offset;
489    e.length = req->length;
490    publish( msgs, &e );
491}
492
493static void
494fireGotRej( tr_peermsgs * msgs, const struct peer_request * req )
495{
496    tr_peer_event e = TR_PEER_EVENT_INIT;
497    e.eventType = TR_PEER_CLIENT_GOT_REJ;
498    e.pieceIndex = req->index;
499    e.offset = req->offset;
500    e.length = req->length;
501    publish( msgs, &e );
502}
503
504static void
505fireGotChoke( tr_peermsgs * msgs )
506{
507    tr_peer_event e = TR_PEER_EVENT_INIT;
508    e.eventType = TR_PEER_CLIENT_GOT_CHOKE;
509    publish( msgs, &e );
510}
511
512static void
513fireClientGotHaveAll( tr_peermsgs * msgs )
514{
515    tr_peer_event e = TR_PEER_EVENT_INIT;
516    e.eventType = TR_PEER_CLIENT_GOT_HAVE_ALL;
517    publish( msgs, &e );
518}
519
520static void
521fireClientGotHaveNone( tr_peermsgs * msgs )
522{
523    tr_peer_event e = TR_PEER_EVENT_INIT;
524    e.eventType = TR_PEER_CLIENT_GOT_HAVE_NONE;
525    publish( msgs, &e );
526}
527
528static void
529fireClientGotData( tr_peermsgs * msgs,
530                   uint32_t      length,
531                   int           wasPieceData )
532{
533    tr_peer_event e = TR_PEER_EVENT_INIT;
534
535    e.length = length;
536    e.eventType = TR_PEER_CLIENT_GOT_DATA;
537    e.wasPieceData = wasPieceData;
538    publish( msgs, &e );
539}
540
541static void
542fireClientGotSuggest( tr_peermsgs * msgs, uint32_t pieceIndex )
543{
544    tr_peer_event e = TR_PEER_EVENT_INIT;
545    e.eventType = TR_PEER_CLIENT_GOT_SUGGEST;
546    e.pieceIndex = pieceIndex;
547    publish( msgs, &e );
548}
549
550static void
551fireClientGotPort( tr_peermsgs * msgs, tr_port port )
552{
553    tr_peer_event e = TR_PEER_EVENT_INIT;
554    e.eventType = TR_PEER_CLIENT_GOT_PORT;
555    e.port = port;
556    publish( msgs, &e );
557}
558
559static void
560fireClientGotAllowedFast( tr_peermsgs * msgs, uint32_t pieceIndex )
561{
562    tr_peer_event e = TR_PEER_EVENT_INIT;
563    e.eventType = TR_PEER_CLIENT_GOT_ALLOWED_FAST;
564    e.pieceIndex = pieceIndex;
565    publish( msgs, &e );
566}
567
568static void
569fireClientGotBitfield( tr_peermsgs * msgs, tr_bitfield * bitfield )
570{
571    tr_peer_event e = TR_PEER_EVENT_INIT;
572    e.eventType = TR_PEER_CLIENT_GOT_BITFIELD;
573    e.bitfield = bitfield;
574    publish( msgs, &e );
575}
576
577static void
578fireClientGotHave( tr_peermsgs * msgs, tr_piece_index_t index )
579{
580    tr_peer_event e = TR_PEER_EVENT_INIT;
581    e.eventType = TR_PEER_CLIENT_GOT_HAVE;
582    e.pieceIndex = index;
583    publish( msgs, &e );
584}
585
586static void
587firePeerGotData( tr_peermsgs  * msgs,
588                 uint32_t       length,
589                 int            wasPieceData )
590{
591    tr_peer_event e = TR_PEER_EVENT_INIT;
592
593    e.length = length;
594    e.eventType = TR_PEER_PEER_GOT_DATA;
595    e.wasPieceData = wasPieceData;
596
597    publish( msgs, &e );
598}
599
600/**
601***  ALLOWED FAST SET
602***  For explanation, see http://www.bittorrent.org/beps/bep_0006.html
603**/
604
605size_t
606tr_generateAllowedSet( tr_piece_index_t * setmePieces,
607                       size_t             desiredSetSize,
608                       size_t             pieceCount,
609                       const uint8_t    * infohash,
610                       const tr_address * addr )
611{
612    size_t setSize = 0;
613
614    assert( setmePieces );
615    assert( desiredSetSize <= pieceCount );
616    assert( desiredSetSize );
617    assert( pieceCount );
618    assert( infohash );
619    assert( addr );
620
621    if( addr->type == TR_AF_INET )
622    {
623        uint8_t w[SHA_DIGEST_LENGTH + 4], *walk=w;
624        uint8_t x[SHA_DIGEST_LENGTH];
625
626        uint32_t ui32 = ntohl( htonl( addr->addr.addr4.s_addr ) & 0xffffff00 );   /* (1) */
627        memcpy( w, &ui32, sizeof( uint32_t ) );
628        walk += sizeof( uint32_t );
629        memcpy( walk, infohash, SHA_DIGEST_LENGTH );                 /* (2) */
630        walk += SHA_DIGEST_LENGTH;
631        tr_sha1( x, w, walk-w, NULL );                               /* (3) */
632        assert( sizeof( w ) == walk-w );
633
634        while( setSize<desiredSetSize )
635        {
636            int i;
637            for( i=0; i<5 && setSize<desiredSetSize; ++i )           /* (4) */
638            {
639                size_t k;
640                uint32_t j = i * 4;                                  /* (5) */
641                uint32_t y = ntohl( *( uint32_t* )( x + j ) );       /* (6) */
642                uint32_t index = y % pieceCount;                     /* (7) */
643
644                for( k=0; k<setSize; ++k )                           /* (8) */
645                    if( setmePieces[k] == index )
646                        break;
647
648                if( k == setSize )
649                    setmePieces[setSize++] = index;                  /* (9) */
650            }
651
652            tr_sha1( x, x, sizeof( x ), NULL );                      /* (3) */
653        }
654    }
655
656    return setSize;
657}
658
659static void
660updateFastSet( tr_peermsgs * msgs UNUSED )
661{
662#if 0
663    const tr_bool fext = tr_peerIoSupportsFEXT( msgs->peer->io );
664    const int peerIsNeedy = msgs->peer->progress < 0.10;
665
666    if( fext && peerIsNeedy && !msgs->haveFastSet )
667    {
668        size_t i;
669        const struct tr_address * addr = tr_peerIoGetAddress( msgs->peer->io, NULL );
670        const tr_info * inf = &msgs->torrent->info;
671        const size_t numwant = MIN( MAX_FAST_SET_SIZE, inf->pieceCount );
672
673        /* build the fast set */
674        msgs->fastsetSize = tr_generateAllowedSet( msgs->fastset, numwant, inf->pieceCount, inf->hash, addr );
675        msgs->haveFastSet = 1;
676
677        /* send it to the peer */
678        for( i=0; i<msgs->fastsetSize; ++i )
679            protocolSendAllowedFast( msgs, msgs->fastset[i] );
680    }
681#endif
682}
683
684/**
685***  INTEREST
686**/
687
688static void
689sendInterest( tr_peermsgs * msgs, tr_bool clientIsInterested )
690{
691    struct evbuffer * out = msgs->outMessages;
692
693    assert( msgs );
694    assert( tr_isBool( clientIsInterested ) );
695
696    msgs->peer->clientIsInterested = clientIsInterested;
697    dbgmsg( msgs, "Sending %s", clientIsInterested ? "Interested" : "Not Interested" );
698    evbuffer_add_uint32( out, sizeof( uint8_t ) );
699    evbuffer_add_uint8 ( out, clientIsInterested ? BT_INTERESTED : BT_NOT_INTERESTED );
700
701    pokeBatchPeriod( msgs, HIGH_PRIORITY_INTERVAL_SECS );
702    dbgOutMessageLen( msgs );
703}
704
705static void
706updateInterest( tr_peermsgs * msgs UNUSED )
707{
708    /* FIXME -- might need to poke the mgr on startup */
709}
710
711void
712tr_peerMsgsSetInterested( tr_peermsgs * msgs, int isInterested )
713{
714    assert( tr_isBool( isInterested ) );
715
716    if( isInterested != msgs->peer->clientIsInterested )
717        sendInterest( msgs, isInterested );
718}
719
720static tr_bool
721popNextMetadataRequest( tr_peermsgs * msgs, int * piece )
722{
723    if( msgs->peerAskedForMetadataCount == 0 )
724        return FALSE;
725
726    *piece = msgs->peerAskedForMetadata[0];
727
728    tr_removeElementFromArray( msgs->peerAskedForMetadata, 0, sizeof( int ),
729                               msgs->peerAskedForMetadataCount-- );
730
731    return TRUE;
732}
733
734static tr_bool
735popNextRequest( tr_peermsgs * msgs, struct peer_request * setme )
736{
737    if( msgs->peer->pendingReqsToClient == 0 )
738        return FALSE;
739
740    *setme = msgs->peerAskedFor[0];
741
742    tr_removeElementFromArray( msgs->peerAskedFor, 0, sizeof( struct peer_request ),
743                               msgs->peer->pendingReqsToClient-- );
744
745    return TRUE;
746}
747
748static void
749cancelAllRequestsToClient( tr_peermsgs * msgs )
750{
751    struct peer_request req;
752    const int mustSendCancel = tr_peerIoSupportsFEXT( msgs->peer->io );
753
754    while( popNextRequest( msgs, &req ))
755        if( mustSendCancel )
756            protocolSendReject( msgs, &req );
757}
758
759void
760tr_peerMsgsSetChoke( tr_peermsgs * msgs,
761                     int           choke )
762{
763    const time_t now = tr_time( );
764    const time_t fibrillationTime = now - MIN_CHOKE_PERIOD_SEC;
765
766    assert( msgs );
767    assert( msgs->peer );
768    assert( choke == 0 || choke == 1 );
769
770    if( msgs->peer->chokeChangedAt > fibrillationTime )
771    {
772        dbgmsg( msgs, "Not changing choke to %d to avoid fibrillation", choke );
773    }
774    else if( msgs->peer->peerIsChoked != choke )
775    {
776        msgs->peer->peerIsChoked = choke;
777        if( choke )
778            cancelAllRequestsToClient( msgs );
779        protocolSendChoke( msgs, choke );
780        msgs->peer->chokeChangedAt = now;
781    }
782}
783
784/**
785***
786**/
787
788void
789tr_peerMsgsHave( tr_peermsgs * msgs,
790                 uint32_t      index )
791{
792    protocolSendHave( msgs, index );
793
794    /* since we have more pieces now, we might not be interested in this peer */
795    updateInterest( msgs );
796}
797
798/**
799***
800**/
801
802static tr_bool
803reqIsValid( const tr_peermsgs * peer,
804            uint32_t            index,
805            uint32_t            offset,
806            uint32_t            length )
807{
808    return tr_torrentReqIsValid( peer->torrent, index, offset, length );
809}
810
811static tr_bool
812requestIsValid( const tr_peermsgs * msgs, const struct peer_request * req )
813{
814    return reqIsValid( msgs, req->index, req->offset, req->length );
815}
816
817void
818tr_peerMsgsCancel( tr_peermsgs * msgs, tr_block_index_t block )
819{
820    struct peer_request req;
821/*fprintf( stderr, "SENDING CANCEL MESSAGE FOR BLOCK %zu\n\t\tFROM PEER %p ------------------------------------\n", (size_t)block, msgs->peer );*/
822    blockToReq( msgs->torrent, block, &req );
823    protocolSendCancel( msgs, &req );
824}
825
826/**
827***
828**/
829
830static void
831sendLtepHandshake( tr_peermsgs * msgs )
832{
833    tr_benc val, *m;
834    char * buf;
835    int len;
836    tr_bool allow_pex;
837    tr_bool allow_metadata_xfer;
838    struct evbuffer * out = msgs->outMessages;
839    const unsigned char * ipv6 = tr_globalIPv6();
840
841    if( msgs->clientSentLtepHandshake )
842        return;
843
844    dbgmsg( msgs, "sending an ltep handshake" );
845    msgs->clientSentLtepHandshake = 1;
846
847    /* decide if we want to advertise metadata xfer support (BEP 9) */
848    if( tr_torrentIsPrivate( msgs->torrent ) )
849        allow_metadata_xfer = 0;
850    else
851        allow_metadata_xfer = 1;
852
853    /* decide if we want to advertise pex support */
854    if( !tr_torrentAllowsPex( msgs->torrent ) )
855        allow_pex = 0;
856    else if( msgs->peerSentLtepHandshake )
857        allow_pex = msgs->peerSupportsPex ? 1 : 0;
858    else
859        allow_pex = 1;
860
861    tr_bencInitDict( &val, 8 );
862    tr_bencDictAddInt( &val, "e", getSession(msgs)->encryptionMode != TR_CLEAR_PREFERRED );
863    if( ipv6 != NULL )
864        tr_bencDictAddRaw( &val, "ipv6", ipv6, 16 );
865    if( allow_metadata_xfer && tr_torrentHasMetadata( msgs->torrent )
866                            && ( msgs->torrent->infoDictLength > 0 ) )
867        tr_bencDictAddInt( &val, "metadata_size", msgs->torrent->infoDictLength );
868    tr_bencDictAddInt( &val, "p", tr_sessionGetPublicPeerPort( getSession(msgs) ) );
869    tr_bencDictAddInt( &val, "reqq", REQQ );
870    tr_bencDictAddInt( &val, "upload_only", tr_torrentIsSeed( msgs->torrent ) );
871    tr_bencDictAddStr( &val, "v", TR_NAME " " USERAGENT_PREFIX );
872    m  = tr_bencDictAddDict( &val, "m", 2 );
873    if( allow_metadata_xfer )
874        tr_bencDictAddInt( m, "ut_metadata", UT_METADATA_ID );
875    if( allow_pex )
876        tr_bencDictAddInt( m, "ut_pex", UT_PEX_ID );
877
878    buf = tr_bencToStr( &val, TR_FMT_BENC, &len );
879
880    evbuffer_add_uint32( out, 2 * sizeof( uint8_t ) + len );
881    evbuffer_add_uint8 ( out, BT_LTEP );
882    evbuffer_add_uint8 ( out, LTEP_HANDSHAKE );
883    evbuffer_add       ( out, buf, len );
884    pokeBatchPeriod( msgs, IMMEDIATE_PRIORITY_INTERVAL_SECS );
885    dbgOutMessageLen( msgs );
886
887    /* cleanup */
888    tr_bencFree( &val );
889    tr_free( buf );
890}
891
892static void
893parseLtepHandshake( tr_peermsgs *     msgs,
894                    int               len,
895                    struct evbuffer * inbuf )
896{
897    int64_t   i;
898    tr_benc   val, * sub;
899    uint8_t * tmp = tr_new( uint8_t, len );
900    const uint8_t *addr;
901    size_t addr_len;
902    tr_pex pex;
903    int8_t seedProbability = -1;
904
905    memset( &pex, 0, sizeof( tr_pex ) );
906
907    tr_peerIoReadBytes( msgs->peer->io, inbuf, tmp, len );
908    msgs->peerSentLtepHandshake = 1;
909
910    if( tr_bencLoad( tmp, len, &val, NULL ) || !tr_bencIsDict( &val ) )
911    {
912        dbgmsg( msgs, "GET  extended-handshake, couldn't get dictionary" );
913        tr_free( tmp );
914        return;
915    }
916
917    dbgmsg( msgs, "here is the handshake: [%*.*s]", len, len,  tmp );
918
919    /* does the peer prefer encrypted connections? */
920    if( tr_bencDictFindInt( &val, "e", &i ) ) {
921        msgs->peer->encryption_preference = i ? ENCRYPTION_PREFERENCE_YES
922                                              : ENCRYPTION_PREFERENCE_NO;
923        if( i )
924            pex.flags |= ADDED_F_ENCRYPTION_FLAG;
925    }
926
927    /* check supported messages for utorrent pex */
928    msgs->peerSupportsPex = 0;
929    msgs->peerSupportsMetadataXfer = 0;
930
931    if( tr_bencDictFindDict( &val, "m", &sub ) ) {
932        if( tr_bencDictFindInt( sub, "ut_pex", &i ) ) {
933            msgs->peerSupportsPex = i != 0;
934            msgs->ut_pex_id = (uint8_t) i;
935            dbgmsg( msgs, "msgs->ut_pex is %d", (int)msgs->ut_pex_id );
936        }
937        if( tr_bencDictFindInt( sub, "ut_metadata", &i ) ) {
938            msgs->peerSupportsMetadataXfer = i != 0;
939            msgs->ut_metadata_id = (uint8_t) i;
940            dbgmsg( msgs, "msgs->ut_metadata_id is %d", (int)msgs->ut_metadata_id );
941        }
942    }
943
944    /* look for metainfo size (BEP 9) */
945    if( tr_bencDictFindInt( &val, "metadata_size", &i ) )
946        tr_torrentSetMetadataSizeHint( msgs->torrent, i );
947
948    /* look for upload_only (BEP 21) */
949    if( tr_bencDictFindInt( &val, "upload_only", &i ) )
950        seedProbability = i==0 ? 0 : 100;
951
952    /* get peer's listening port */
953    if( tr_bencDictFindInt( &val, "p", &i ) ) {
954        pex.port = htons( (uint16_t)i );
955        fireClientGotPort( msgs, pex.port );
956        dbgmsg( msgs, "peer's port is now %d", (int)i );
957    }
958
959    if( tr_peerIoIsIncoming( msgs->peer->io )
960        && tr_bencDictFindRaw( &val, "ipv4", &addr, &addr_len )
961        && ( addr_len == 4 ) )
962    {
963        pex.addr.type = TR_AF_INET;
964        memcpy( &pex.addr.addr.addr4, addr, 4 );
965        tr_peerMgrAddPex( msgs->torrent, TR_PEER_FROM_LTEP, &pex, seedProbability );
966    }
967
968    if( tr_peerIoIsIncoming( msgs->peer->io )
969        && tr_bencDictFindRaw( &val, "ipv6", &addr, &addr_len )
970        && ( addr_len == 16 ) )
971    {
972        pex.addr.type = TR_AF_INET6;
973        memcpy( &pex.addr.addr.addr6, addr, 16 );
974        tr_peerMgrAddPex( msgs->torrent, TR_PEER_FROM_LTEP, &pex, seedProbability );
975    }
976
977    /* get peer's maximum request queue size */
978    if( tr_bencDictFindInt( &val, "reqq", &i ) )
979        msgs->reqq = i;
980
981    tr_bencFree( &val );
982    tr_free( tmp );
983}
984
985static void
986parseUtMetadata( tr_peermsgs * msgs, int msglen, struct evbuffer * inbuf )
987{
988    tr_benc dict;
989    char * msg_end;
990    char * benc_end;
991    int64_t msg_type = -1;
992    int64_t piece = -1;
993    int64_t total_size = 0;
994    uint8_t * tmp = tr_new( uint8_t, msglen );
995
996    tr_peerIoReadBytes( msgs->peer->io, inbuf, tmp, msglen );
997    msg_end = (char*)tmp + msglen;
998
999    if( !tr_bencLoad( tmp, msglen, &dict, &benc_end ) )
1000    {
1001        tr_bencDictFindInt( &dict, "msg_type", &msg_type );
1002        tr_bencDictFindInt( &dict, "piece", &piece );
1003        tr_bencDictFindInt( &dict, "total_size", &total_size );
1004        tr_bencFree( &dict );
1005    }
1006
1007    dbgmsg( msgs, "got ut_metadata msg: type %d, piece %d, total_size %d",
1008            (int)msg_type, (int)piece, (int)total_size );
1009
1010    if( msg_type == METADATA_MSG_TYPE_REJECT )
1011    {
1012        /* NOOP */
1013    }
1014
1015    if( ( msg_type == METADATA_MSG_TYPE_DATA )
1016        && ( !tr_torrentHasMetadata( msgs->torrent ) )
1017        && ( msg_end - benc_end <= METADATA_PIECE_SIZE )
1018        && ( piece * METADATA_PIECE_SIZE + (msg_end - benc_end) <= total_size ) )
1019    {
1020        const int pieceLen = msg_end - benc_end;
1021        tr_torrentSetMetadataPiece( msgs->torrent, piece, benc_end, pieceLen );
1022    }
1023
1024    if( msg_type == METADATA_MSG_TYPE_REQUEST )
1025    {
1026        if( ( piece >= 0 )
1027            && tr_torrentHasMetadata( msgs->torrent )
1028            && !tr_torrentIsPrivate( msgs->torrent )
1029            && ( msgs->peerAskedForMetadataCount < METADATA_REQQ ) )
1030        {
1031            msgs->peerAskedForMetadata[msgs->peerAskedForMetadataCount++] = piece;
1032        }
1033        else
1034        {
1035            tr_benc tmp;
1036            int payloadLen;
1037            char * payload;
1038            struct evbuffer * out = msgs->outMessages;
1039
1040            /* build the rejection message */
1041            tr_bencInitDict( &tmp, 2 );
1042            tr_bencDictAddInt( &tmp, "msg_type", METADATA_MSG_TYPE_REJECT );
1043            tr_bencDictAddInt( &tmp, "piece", piece );
1044            payload = tr_bencToStr( &tmp, TR_FMT_BENC, &payloadLen );
1045            tr_bencFree( &tmp );
1046
1047            /* write it out as a LTEP message to our outMessages buffer */
1048            evbuffer_add_uint32( out, 2 * sizeof( uint8_t ) + payloadLen );
1049            evbuffer_add_uint8 ( out, BT_LTEP );
1050            evbuffer_add_uint8 ( out, msgs->ut_metadata_id );
1051            evbuffer_add       ( out, payload, payloadLen );
1052            pokeBatchPeriod( msgs, HIGH_PRIORITY_INTERVAL_SECS );
1053            dbgOutMessageLen( msgs );
1054
1055            tr_free( payload );
1056        }
1057    }
1058
1059    tr_free( tmp );
1060}
1061
1062static void
1063parseUtPex( tr_peermsgs * msgs, int msglen, struct evbuffer * inbuf )
1064{
1065    int loaded = 0;
1066    uint8_t * tmp = tr_new( uint8_t, msglen );
1067    tr_benc val;
1068    tr_torrent * tor = msgs->torrent;
1069    const uint8_t * added;
1070    size_t added_len;
1071
1072    tr_peerIoReadBytes( msgs->peer->io, inbuf, tmp, msglen );
1073
1074    if( tr_torrentAllowsPex( tor )
1075      && ( ( loaded = !tr_bencLoad( tmp, msglen, &val, NULL ) ) ) )
1076    {
1077        if( tr_bencDictFindRaw( &val, "added", &added, &added_len ) )
1078        {
1079            tr_pex * pex;
1080            size_t i, n;
1081            size_t added_f_len = 0;
1082            const uint8_t * added_f = NULL;
1083
1084            tr_bencDictFindRaw( &val, "added.f", &added_f, &added_f_len );
1085            pex = tr_peerMgrCompactToPex( added, added_len, added_f, added_f_len, &n );
1086
1087            n = MIN( n, MAX_PEX_PEER_COUNT );
1088            for( i=0; i<n; ++i )
1089            {
1090                int seedProbability = -1;
1091                if( i < added_f_len ) seedProbability = ( added_f[i] & ADDED_F_SEED_FLAG ) ? 100 : 0;
1092                tr_peerMgrAddPex( tor, TR_PEER_FROM_PEX, pex+i, seedProbability );
1093            }
1094
1095            tr_free( pex );
1096        }
1097
1098        if( tr_bencDictFindRaw( &val, "added6", &added, &added_len ) )
1099        {
1100            tr_pex * pex;
1101            size_t i, n;
1102            size_t added_f_len = 0;
1103            const uint8_t * added_f = NULL;
1104
1105            tr_bencDictFindRaw( &val, "added6.f", &added_f, &added_f_len );
1106            pex = tr_peerMgrCompact6ToPex( added, added_len, added_f, added_f_len, &n );
1107
1108            n = MIN( n, MAX_PEX_PEER_COUNT );
1109            for( i=0; i<n; ++i )
1110            {
1111                int seedProbability = -1;
1112                if( i < added_f_len ) seedProbability = ( added_f[i] & ADDED_F_SEED_FLAG ) ? 100 : 0;
1113                tr_peerMgrAddPex( tor, TR_PEER_FROM_PEX, pex+i, seedProbability );
1114            }
1115
1116            tr_free( pex );
1117        }
1118    }
1119
1120    if( loaded )
1121        tr_bencFree( &val );
1122    tr_free( tmp );
1123}
1124
1125static void sendPex( tr_peermsgs * msgs );
1126
1127static void
1128parseLtep( tr_peermsgs * msgs, int msglen, struct evbuffer  * inbuf )
1129{
1130    uint8_t ltep_msgid;
1131
1132    tr_peerIoReadUint8( msgs->peer->io, inbuf, &ltep_msgid );
1133    msglen--;
1134
1135    if( ltep_msgid == LTEP_HANDSHAKE )
1136    {
1137        dbgmsg( msgs, "got ltep handshake" );
1138        parseLtepHandshake( msgs, msglen, inbuf );
1139        if( tr_peerIoSupportsLTEP( msgs->peer->io ) )
1140        {
1141            sendLtepHandshake( msgs );
1142            sendPex( msgs );
1143        }
1144    }
1145    else if( ltep_msgid == UT_PEX_ID )
1146    {
1147        dbgmsg( msgs, "got ut pex" );
1148        msgs->peerSupportsPex = 1;
1149        parseUtPex( msgs, msglen, inbuf );
1150    }
1151    else if( ltep_msgid == UT_METADATA_ID )
1152    {
1153        dbgmsg( msgs, "got ut metadata" );
1154        msgs->peerSupportsMetadataXfer = 1;
1155        parseUtMetadata( msgs, msglen, inbuf );
1156    }
1157    else
1158    {
1159        dbgmsg( msgs, "skipping unknown ltep message (%d)", (int)ltep_msgid );
1160        evbuffer_drain( inbuf, msglen );
1161    }
1162}
1163
1164static int
1165readBtLength( tr_peermsgs *     msgs,
1166              struct evbuffer * inbuf,
1167              size_t            inlen )
1168{
1169    uint32_t len;
1170
1171    if( inlen < sizeof( len ) )
1172        return READ_LATER;
1173
1174    tr_peerIoReadUint32( msgs->peer->io, inbuf, &len );
1175
1176    if( len == 0 ) /* peer sent us a keepalive message */
1177        dbgmsg( msgs, "got KeepAlive" );
1178    else
1179    {
1180        msgs->incoming.length = len;
1181        msgs->state = AWAITING_BT_ID;
1182    }
1183
1184    return READ_NOW;
1185}
1186
1187static int readBtMessage( tr_peermsgs     * msgs,
1188                          struct evbuffer * inbuf,
1189                          size_t            inlen );
1190
1191static int
1192readBtId( tr_peermsgs * msgs, struct evbuffer  * inbuf, size_t inlen )
1193{
1194    uint8_t id;
1195
1196    if( inlen < sizeof( uint8_t ) )
1197        return READ_LATER;
1198
1199    tr_peerIoReadUint8( msgs->peer->io, inbuf, &id );
1200    msgs->incoming.id = id;
1201    dbgmsg( msgs, "msgs->incoming.id is now %d; msgs->incoming.length is %zu", id, (size_t)msgs->incoming.length );
1202
1203    if( id == BT_PIECE )
1204    {
1205        msgs->state = AWAITING_BT_PIECE;
1206        return READ_NOW;
1207    }
1208    else if( msgs->incoming.length != 1 )
1209    {
1210        msgs->state = AWAITING_BT_MESSAGE;
1211        return READ_NOW;
1212    }
1213    else return readBtMessage( msgs, inbuf, inlen - 1 );
1214}
1215
1216static void
1217updatePeerProgress( tr_peermsgs * msgs )
1218{
1219    msgs->peer->progress = tr_bitsetPercent( &msgs->peer->have );
1220    dbgmsg( msgs, "peer progress is %f", msgs->peer->progress );
1221    updateFastSet( msgs );
1222    updateInterest( msgs );
1223    firePeerProgress( msgs );
1224}
1225
1226static void
1227prefetchPieces( tr_peermsgs *msgs )
1228{
1229    int i;
1230
1231    if( !getSession(msgs)->isPrefetchEnabled )
1232        return;
1233
1234    /* Maintain 12 prefetched blocks per unchoked peer */
1235    for( i=msgs->prefetchCount; i<msgs->peer->pendingReqsToClient && i<12; ++i )
1236    {
1237        const struct peer_request * req = msgs->peerAskedFor + i;
1238        if( requestIsValid( msgs, req ) )
1239        {
1240            tr_cachePrefetchBlock( getSession(msgs)->cache, msgs->torrent, req->index, req->offset, req->length );
1241            ++msgs->prefetchCount;
1242        }
1243    }
1244}
1245
1246static void
1247peerMadeRequest( tr_peermsgs *               msgs,
1248                 const struct peer_request * req )
1249{
1250    const tr_bool fext = tr_peerIoSupportsFEXT( msgs->peer->io );
1251    const int reqIsValid = requestIsValid( msgs, req );
1252    const int clientHasPiece = reqIsValid && tr_cpPieceIsComplete( &msgs->torrent->completion, req->index );
1253    const int peerIsChoked = msgs->peer->peerIsChoked;
1254
1255    int allow = FALSE;
1256
1257    if( !reqIsValid )
1258        dbgmsg( msgs, "rejecting an invalid request." );
1259    else if( !clientHasPiece )
1260        dbgmsg( msgs, "rejecting request for a piece we don't have." );
1261    else if( peerIsChoked )
1262        dbgmsg( msgs, "rejecting request from choked peer" );
1263    else if( msgs->peer->pendingReqsToClient + 1 >= REQQ )
1264        dbgmsg( msgs, "rejecting request ... reqq is full" );
1265    else
1266        allow = TRUE;
1267
1268    if( allow ) {
1269        msgs->peerAskedFor[msgs->peer->pendingReqsToClient++] = *req;
1270        prefetchPieces( msgs );
1271    } else if( fext ) {
1272        protocolSendReject( msgs, req );
1273    }
1274}
1275
1276static tr_bool
1277messageLengthIsCorrect( const tr_peermsgs * msg, uint8_t id, uint32_t len )
1278{
1279    switch( id )
1280    {
1281        case BT_CHOKE:
1282        case BT_UNCHOKE:
1283        case BT_INTERESTED:
1284        case BT_NOT_INTERESTED:
1285        case BT_FEXT_HAVE_ALL:
1286        case BT_FEXT_HAVE_NONE:
1287            return len == 1;
1288
1289        case BT_HAVE:
1290        case BT_FEXT_SUGGEST:
1291        case BT_FEXT_ALLOWED_FAST:
1292            return len == 5;
1293
1294        case BT_BITFIELD:
1295            return len == ( msg->torrent->info.pieceCount + 7u ) / 8u + 1u;
1296
1297        case BT_REQUEST:
1298        case BT_CANCEL:
1299        case BT_FEXT_REJECT:
1300            return len == 13;
1301
1302        case BT_PIECE:
1303            return len > 9 && len <= 16393;
1304
1305        case BT_PORT:
1306            return len == 3;
1307
1308        case BT_LTEP:
1309            return len >= 2;
1310
1311        default:
1312            return FALSE;
1313    }
1314}
1315
1316static int clientGotBlock( tr_peermsgs *               msgs,
1317                           struct evbuffer *           block,
1318                           const struct peer_request * req );
1319
1320static int
1321readBtPiece( tr_peermsgs      * msgs,
1322             struct evbuffer  * inbuf,
1323             size_t             inlen,
1324             size_t           * setme_piece_bytes_read )
1325{
1326    struct peer_request * req = &msgs->incoming.blockReq;
1327
1328    assert( evbuffer_get_length( inbuf ) >= inlen );
1329    dbgmsg( msgs, "In readBtPiece" );
1330
1331    if( !req->length )
1332    {
1333        if( inlen < 8 )
1334            return READ_LATER;
1335
1336        tr_peerIoReadUint32( msgs->peer->io, inbuf, &req->index );
1337        tr_peerIoReadUint32( msgs->peer->io, inbuf, &req->offset );
1338        req->length = msgs->incoming.length - 9;
1339        dbgmsg( msgs, "got incoming block header %u:%u->%u", req->index, req->offset, req->length );
1340        return READ_NOW;
1341    }
1342    else
1343    {
1344        int err;
1345
1346        /* read in another chunk of data */
1347        const size_t nLeft = req->length - evbuffer_get_length( msgs->incoming.block );
1348        size_t n = MIN( nLeft, inlen );
1349        size_t i = n;
1350        void * buf = tr_sessionGetBuffer( getSession( msgs ) );
1351        const size_t buflen = SESSION_BUFFER_SIZE;
1352
1353        while( i > 0 )
1354        {
1355            const size_t thisPass = MIN( i, buflen );
1356            tr_peerIoReadBytes( msgs->peer->io, inbuf, buf, thisPass );
1357            evbuffer_add( msgs->incoming.block, buf, thisPass );
1358            i -= thisPass;
1359        }
1360
1361        tr_sessionReleaseBuffer( getSession( msgs ) );
1362        buf = NULL;
1363
1364        fireClientGotData( msgs, n, TRUE );
1365        *setme_piece_bytes_read += n;
1366        dbgmsg( msgs, "got %zu bytes for block %u:%u->%u ... %d remain",
1367               n, req->index, req->offset, req->length,
1368               (int)( req->length - evbuffer_get_length( msgs->incoming.block ) ) );
1369        if( evbuffer_get_length( msgs->incoming.block ) < req->length )
1370            return READ_LATER;
1371
1372        /* we've got the whole block ... process it */
1373        err = clientGotBlock( msgs, msgs->incoming.block, req );
1374
1375        /* cleanup */
1376        evbuffer_free( msgs->incoming.block );
1377        msgs->incoming.block = evbuffer_new( );
1378        req->length = 0;
1379        msgs->state = AWAITING_BT_LENGTH;
1380        return err ? READ_ERR : READ_NOW;
1381    }
1382}
1383
1384static void updateDesiredRequestCount( tr_peermsgs * msgs );
1385
1386static int
1387readBtMessage( tr_peermsgs * msgs, struct evbuffer * inbuf, size_t inlen )
1388{
1389    uint32_t      ui32;
1390    uint32_t      msglen = msgs->incoming.length;
1391    const uint8_t id = msgs->incoming.id;
1392#ifndef NDEBUG
1393    const size_t  startBufLen = evbuffer_get_length( inbuf );
1394#endif
1395    const tr_bool fext = tr_peerIoSupportsFEXT( msgs->peer->io );
1396
1397    --msglen; /* id length */
1398
1399    dbgmsg( msgs, "got BT id %d, len %d, buffer size is %zu", (int)id, (int)msglen, inlen );
1400
1401    if( inlen < msglen )
1402        return READ_LATER;
1403
1404    if( !messageLengthIsCorrect( msgs, id, msglen + 1 ) )
1405    {
1406        dbgmsg( msgs, "bad packet - BT message #%d with a length of %d", (int)id, (int)msglen );
1407        fireError( msgs, EMSGSIZE );
1408        return READ_ERR;
1409    }
1410
1411    switch( id )
1412    {
1413        case BT_CHOKE:
1414            dbgmsg( msgs, "got Choke" );
1415            msgs->peer->clientIsChoked = 1;
1416            if( !fext )
1417                fireGotChoke( msgs );
1418            break;
1419
1420        case BT_UNCHOKE:
1421            dbgmsg( msgs, "got Unchoke" );
1422            msgs->peer->clientIsChoked = 0;
1423            updateDesiredRequestCount( msgs );
1424            break;
1425
1426        case BT_INTERESTED:
1427            dbgmsg( msgs, "got Interested" );
1428            msgs->peer->peerIsInterested = 1;
1429            break;
1430
1431        case BT_NOT_INTERESTED:
1432            dbgmsg( msgs, "got Not Interested" );
1433            msgs->peer->peerIsInterested = 0;
1434            break;
1435
1436        case BT_HAVE:
1437            tr_peerIoReadUint32( msgs->peer->io, inbuf, &ui32 );
1438            dbgmsg( msgs, "got Have: %u", ui32 );
1439            if( tr_torrentHasMetadata( msgs->torrent )
1440                    && ( ui32 >= msgs->torrent->info.pieceCount ) )
1441            {
1442                fireError( msgs, ERANGE );
1443                return READ_ERR;
1444            }
1445
1446            /* a peer can send the same HAVE message twice... */
1447            if( !tr_bitsetHas( &msgs->peer->have, ui32 ) )
1448                if( !tr_bitsetAdd( &msgs->peer->have, ui32 ) )
1449                    fireClientGotHave( msgs, ui32 );
1450            updatePeerProgress( msgs );
1451            break;
1452
1453        case BT_BITFIELD: {
1454            const size_t bitCount = tr_torrentHasMetadata( msgs->torrent )
1455                                  ? msgs->torrent->info.pieceCount
1456                                  : msglen * 8;
1457            dbgmsg( msgs, "got a bitfield" );
1458            tr_bitsetReserve( &msgs->peer->have, bitCount );
1459            tr_peerIoReadBytes( msgs->peer->io, inbuf,
1460                                msgs->peer->have.bitfield.bits, msglen );
1461            fireClientGotBitfield( msgs, &msgs->peer->have.bitfield );
1462            updatePeerProgress( msgs );
1463            break;
1464        }
1465
1466        case BT_REQUEST:
1467        {
1468            struct peer_request r;
1469            tr_peerIoReadUint32( msgs->peer->io, inbuf, &r.index );
1470            tr_peerIoReadUint32( msgs->peer->io, inbuf, &r.offset );
1471            tr_peerIoReadUint32( msgs->peer->io, inbuf, &r.length );
1472            dbgmsg( msgs, "got Request: %u:%u->%u", r.index, r.offset, r.length );
1473            peerMadeRequest( msgs, &r );
1474            break;
1475        }
1476
1477        case BT_CANCEL:
1478        {
1479            int i;
1480            struct peer_request r;
1481            tr_peerIoReadUint32( msgs->peer->io, inbuf, &r.index );
1482            tr_peerIoReadUint32( msgs->peer->io, inbuf, &r.offset );
1483            tr_peerIoReadUint32( msgs->peer->io, inbuf, &r.length );
1484            tr_historyAdd( msgs->peer->cancelsSentToClient, tr_time( ), 1 );
1485            dbgmsg( msgs, "got a Cancel %u:%u->%u", r.index, r.offset, r.length );
1486
1487            for( i=0; i<msgs->peer->pendingReqsToClient; ++i ) {
1488                const struct peer_request * req = msgs->peerAskedFor + i;
1489                if( ( req->index == r.index ) && ( req->offset == r.offset ) && ( req->length == r.length ) )
1490                    break;
1491            }
1492
1493            if( i < msgs->peer->pendingReqsToClient )
1494                tr_removeElementFromArray( msgs->peerAskedFor, i, sizeof( struct peer_request ),
1495                                           msgs->peer->pendingReqsToClient-- );
1496            break;
1497        }
1498
1499        case BT_PIECE:
1500            assert( 0 ); /* handled elsewhere! */
1501            break;
1502
1503        case BT_PORT:
1504            dbgmsg( msgs, "Got a BT_PORT" );
1505            tr_peerIoReadUint16( msgs->peer->io, inbuf, &msgs->peer->dht_port );
1506            if( msgs->peer->dht_port > 0 )
1507                tr_dhtAddNode( getSession(msgs),
1508                               tr_peerAddress( msgs->peer ),
1509                               msgs->peer->dht_port, 0 );
1510            break;
1511
1512        case BT_FEXT_SUGGEST:
1513            dbgmsg( msgs, "Got a BT_FEXT_SUGGEST" );
1514            tr_peerIoReadUint32( msgs->peer->io, inbuf, &ui32 );
1515            if( fext )
1516                fireClientGotSuggest( msgs, ui32 );
1517            else {
1518                fireError( msgs, EMSGSIZE );
1519                return READ_ERR;
1520            }
1521            break;
1522
1523        case BT_FEXT_ALLOWED_FAST:
1524            dbgmsg( msgs, "Got a BT_FEXT_ALLOWED_FAST" );
1525            tr_peerIoReadUint32( msgs->peer->io, inbuf, &ui32 );
1526            if( fext )
1527                fireClientGotAllowedFast( msgs, ui32 );
1528            else {
1529                fireError( msgs, EMSGSIZE );
1530                return READ_ERR;
1531            }
1532            break;
1533
1534        case BT_FEXT_HAVE_ALL:
1535            dbgmsg( msgs, "Got a BT_FEXT_HAVE_ALL" );
1536            if( fext ) {
1537                tr_bitsetSetHaveAll( &msgs->peer->have );
1538                fireClientGotHaveAll( msgs );
1539                updatePeerProgress( msgs );
1540            } else {
1541                fireError( msgs, EMSGSIZE );
1542                return READ_ERR;
1543            }
1544            break;
1545
1546        case BT_FEXT_HAVE_NONE:
1547            dbgmsg( msgs, "Got a BT_FEXT_HAVE_NONE" );
1548            if( fext ) {
1549                tr_bitsetSetHaveNone( &msgs->peer->have );
1550                fireClientGotHaveNone( msgs );
1551                updatePeerProgress( msgs );
1552            } else {
1553                fireError( msgs, EMSGSIZE );
1554                return READ_ERR;
1555            }
1556            break;
1557
1558        case BT_FEXT_REJECT:
1559        {
1560            struct peer_request r;
1561            dbgmsg( msgs, "Got a BT_FEXT_REJECT" );
1562            tr_peerIoReadUint32( msgs->peer->io, inbuf, &r.index );
1563            tr_peerIoReadUint32( msgs->peer->io, inbuf, &r.offset );
1564            tr_peerIoReadUint32( msgs->peer->io, inbuf, &r.length );
1565            if( fext )
1566                fireGotRej( msgs, &r );
1567            else {
1568                fireError( msgs, EMSGSIZE );
1569                return READ_ERR;
1570            }
1571            break;
1572        }
1573
1574        case BT_LTEP:
1575            dbgmsg( msgs, "Got a BT_LTEP" );
1576            parseLtep( msgs, msglen, inbuf );
1577            break;
1578
1579        default:
1580            dbgmsg( msgs, "peer sent us an UNKNOWN: %d", (int)id );
1581            tr_peerIoDrain( msgs->peer->io, inbuf, msglen );
1582            break;
1583    }
1584
1585    assert( msglen + 1 == msgs->incoming.length );
1586    assert( evbuffer_get_length( inbuf ) == startBufLen - msglen );
1587
1588    msgs->state = AWAITING_BT_LENGTH;
1589    return READ_NOW;
1590}
1591
1592static void
1593addPeerToBlamefield( tr_peermsgs * msgs, uint32_t index )
1594{
1595    if( !msgs->peer->blame )
1596         msgs->peer->blame = tr_bitfieldNew( msgs->torrent->info.pieceCount );
1597    tr_bitfieldAdd( msgs->peer->blame, index );
1598}
1599
1600/* returns 0 on success, or an errno on failure */
1601static int
1602clientGotBlock( tr_peermsgs *               msgs,
1603                struct evbuffer *           data,
1604                const struct peer_request * req )
1605{
1606    int err;
1607    tr_torrent * tor = msgs->torrent;
1608    const tr_block_index_t block = _tr_block( tor, req->index, req->offset );
1609
1610    assert( msgs );
1611    assert( req );
1612
1613    if( req->length != tr_torBlockCountBytes( msgs->torrent, block ) ) {
1614        dbgmsg( msgs, "wrong block size -- expected %u, got %d",
1615                tr_torBlockCountBytes( msgs->torrent, block ), req->length );
1616        return EMSGSIZE;
1617    }
1618
1619    dbgmsg( msgs, "got block %u:%u->%u", req->index, req->offset, req->length );
1620
1621    if( !tr_peerMgrDidPeerRequest( msgs->torrent, msgs->peer, block ) ) {
1622        dbgmsg( msgs, "we didn't ask for this message..." );
1623        return 0;
1624    }
1625    if( tr_cpPieceIsComplete( &msgs->torrent->completion, req->index ) ) {
1626        dbgmsg( msgs, "we did ask for this message, but the piece is already complete..." );
1627        return 0;
1628    }
1629
1630    /**
1631    ***  Save the block
1632    **/
1633
1634    if(( err = tr_cacheWriteBlock( getSession(msgs)->cache, tor, req->index, req->offset, req->length, data )))
1635        return err;
1636
1637    addPeerToBlamefield( msgs, req->index );
1638    fireGotBlock( msgs, req );
1639    return 0;
1640}
1641
1642static int peerPulse( void * vmsgs );
1643
1644static void
1645didWrite( tr_peerIo * io UNUSED, size_t bytesWritten, int wasPieceData, void * vmsgs )
1646{
1647    tr_peermsgs * msgs = vmsgs;
1648    firePeerGotData( msgs, bytesWritten, wasPieceData );
1649
1650    if ( tr_isPeerIo( io ) && io->userData )
1651        peerPulse( msgs );
1652}
1653
1654static ReadState
1655canRead( tr_peerIo * io, void * vmsgs, size_t * piece )
1656{
1657    ReadState         ret;
1658    tr_peermsgs *     msgs = vmsgs;
1659    struct evbuffer * in = tr_peerIoGetReadBuffer( io );
1660    const size_t      inlen = evbuffer_get_length( in );
1661
1662    dbgmsg( msgs, "canRead: inlen is %zu, msgs->state is %d", inlen, msgs->state );
1663
1664    if( !inlen )
1665    {
1666        ret = READ_LATER;
1667    }
1668    else if( msgs->state == AWAITING_BT_PIECE )
1669    {
1670        ret = inlen ? readBtPiece( msgs, in, inlen, piece ) : READ_LATER;
1671    }
1672    else switch( msgs->state )
1673    {
1674        case AWAITING_BT_LENGTH:
1675            ret = readBtLength ( msgs, in, inlen ); break;
1676
1677        case AWAITING_BT_ID:
1678            ret = readBtId     ( msgs, in, inlen ); break;
1679
1680        case AWAITING_BT_MESSAGE:
1681            ret = readBtMessage( msgs, in, inlen ); break;
1682
1683        default:
1684            ret = READ_ERR;
1685            assert( 0 );
1686    }
1687
1688    dbgmsg( msgs, "canRead: ret is %d", (int)ret );
1689
1690    /* log the raw data that was read */
1691    if( ( ret != READ_ERR ) && ( evbuffer_get_length( in ) != inlen ) )
1692        fireClientGotData( msgs, inlen - evbuffer_get_length( in ), FALSE );
1693
1694    return ret;
1695}
1696
1697int
1698tr_peerMsgsIsReadingBlock( const tr_peermsgs * msgs, tr_block_index_t block )
1699{
1700    if( msgs->state != AWAITING_BT_PIECE )
1701        return FALSE;
1702
1703    return block == _tr_block( msgs->torrent,
1704                               msgs->incoming.blockReq.index,
1705                               msgs->incoming.blockReq.offset );
1706}
1707
1708/**
1709***
1710**/
1711
1712static void
1713updateDesiredRequestCount( tr_peermsgs * msgs )
1714{
1715    const tr_torrent * const torrent = msgs->torrent;
1716
1717    if( tr_torrentIsSeed( msgs->torrent ) )
1718    {
1719        msgs->desiredRequestCount = 0;
1720    }
1721    else if( msgs->peer->clientIsChoked )
1722    {
1723        msgs->desiredRequestCount = 0;
1724    }
1725    else if( !msgs->peer->clientIsInterested )
1726    {
1727        msgs->desiredRequestCount = 0;
1728    }
1729    else
1730    {
1731        int estimatedBlocksInPeriod;
1732        int rate_Bps;
1733        int irate_Bps;
1734        const int floor = 4;
1735        const int seconds = REQUEST_BUF_SECS;
1736        const uint64_t now = tr_time_msec( );
1737
1738        /* Get the rate limit we should use.
1739         * FIXME: this needs to consider all the other peers as well... */
1740        rate_Bps = tr_peerGetPieceSpeed_Bps( msgs->peer, now, TR_PEER_TO_CLIENT );
1741        if( tr_torrentUsesSpeedLimit( torrent, TR_PEER_TO_CLIENT ) )
1742            rate_Bps = MIN( rate_Bps, tr_torrentGetSpeedLimit_Bps( torrent, TR_PEER_TO_CLIENT ) );
1743
1744        /* honor the session limits, if enabled */
1745        if( tr_torrentUsesSessionLimits( torrent ) )
1746            if( tr_sessionGetActiveSpeedLimit_Bps( torrent->session, TR_PEER_TO_CLIENT, &irate_Bps ) )
1747                rate_Bps = MIN( rate_Bps, irate_Bps );
1748
1749        /* use this desired rate to figure out how
1750         * many requests we should send to this peer */
1751        estimatedBlocksInPeriod = ( rate_Bps * seconds ) / torrent->blockSize;
1752        msgs->desiredRequestCount = MAX( floor, estimatedBlocksInPeriod );
1753
1754        /* honor the peer's maximum request count, if specified */
1755        if( msgs->reqq > 0 )
1756            if( msgs->desiredRequestCount > msgs->reqq )
1757                msgs->desiredRequestCount = msgs->reqq;
1758    }
1759}
1760
1761static void
1762updateMetadataRequests( tr_peermsgs * msgs, time_t now )
1763{
1764    int piece;
1765
1766    if( msgs->peerSupportsMetadataXfer
1767        && tr_torrentGetNextMetadataRequest( msgs->torrent, now, &piece ) )
1768    {
1769        tr_benc tmp;
1770        int payloadLen;
1771        char * payload;
1772        struct evbuffer * out = msgs->outMessages;
1773
1774        /* build the data message */
1775        tr_bencInitDict( &tmp, 3 );
1776        tr_bencDictAddInt( &tmp, "msg_type", METADATA_MSG_TYPE_REQUEST );
1777        tr_bencDictAddInt( &tmp, "piece", piece );
1778        payload = tr_bencToStr( &tmp, TR_FMT_BENC, &payloadLen );
1779        tr_bencFree( &tmp );
1780
1781        dbgmsg( msgs, "requesting metadata piece #%d", piece );
1782
1783        /* write it out as a LTEP message to our outMessages buffer */
1784        evbuffer_add_uint32( out, 2 * sizeof( uint8_t ) + payloadLen );
1785        evbuffer_add_uint8 ( out, BT_LTEP );
1786        evbuffer_add_uint8 ( out, msgs->ut_metadata_id );
1787        evbuffer_add       ( out, payload, payloadLen );
1788        pokeBatchPeriod( msgs, HIGH_PRIORITY_INTERVAL_SECS );
1789        dbgOutMessageLen( msgs );
1790
1791        tr_free( payload );
1792    }
1793}
1794
1795static void
1796updateBlockRequests( tr_peermsgs * msgs )
1797{
1798    if( tr_torrentIsPieceTransferAllowed( msgs->torrent, TR_PEER_TO_CLIENT )
1799        && ( msgs->desiredRequestCount > 0 )
1800        && ( msgs->peer->pendingReqsToPeer <= ( msgs->desiredRequestCount * 0.66 ) ) )
1801    {
1802        int i;
1803        int n;
1804        const int numwant = msgs->desiredRequestCount - msgs->peer->pendingReqsToPeer;
1805        tr_block_index_t * blocks = tr_new( tr_block_index_t, numwant );
1806
1807        tr_peerMgrGetNextRequests( msgs->torrent, msgs->peer, numwant, blocks, &n );
1808
1809        for( i=0; i<n; ++i )
1810        {
1811            struct peer_request req;
1812            blockToReq( msgs->torrent, blocks[i], &req );
1813            protocolSendRequest( msgs, &req );
1814        }
1815
1816        tr_free( blocks );
1817    }
1818}
1819
1820static size_t
1821fillOutputBuffer( tr_peermsgs * msgs, time_t now )
1822{
1823    int piece;
1824    size_t bytesWritten = 0;
1825    struct peer_request req;
1826    const tr_bool haveMessages = evbuffer_get_length( msgs->outMessages ) != 0;
1827    const tr_bool fext = tr_peerIoSupportsFEXT( msgs->peer->io );
1828
1829    /**
1830    ***  Protocol messages
1831    **/
1832
1833    if( haveMessages && !msgs->outMessagesBatchedAt ) /* fresh batch */
1834    {
1835        dbgmsg( msgs, "started an outMessages batch (length is %zu)", evbuffer_get_length( msgs->outMessages ) );
1836        msgs->outMessagesBatchedAt = now;
1837    }
1838    else if( haveMessages && ( ( now - msgs->outMessagesBatchedAt ) >= msgs->outMessagesBatchPeriod ) )
1839    {
1840        const size_t len = evbuffer_get_length( msgs->outMessages );
1841        /* flush the protocol messages */
1842        dbgmsg( msgs, "flushing outMessages... to %p (length is %zu)", msgs->peer->io, len );
1843        tr_peerIoWriteBuf( msgs->peer->io, msgs->outMessages, FALSE );
1844        msgs->clientSentAnythingAt = now;
1845        msgs->outMessagesBatchedAt = 0;
1846        msgs->outMessagesBatchPeriod = LOW_PRIORITY_INTERVAL_SECS;
1847        bytesWritten +=  len;
1848    }
1849
1850    /**
1851    ***  Metadata Pieces
1852    **/
1853
1854    if( ( tr_peerIoGetWriteBufferSpace( msgs->peer->io, now ) >= METADATA_PIECE_SIZE )
1855        && popNextMetadataRequest( msgs, &piece ) )
1856    {
1857        char * data;
1858        int dataLen;
1859        tr_bool ok = FALSE;
1860
1861        data = tr_torrentGetMetadataPiece( msgs->torrent, piece, &dataLen );
1862        if( ( dataLen > 0 ) && ( data != NULL ) )
1863        {
1864            tr_benc tmp;
1865            int payloadLen;
1866            char * payload;
1867            struct evbuffer * out = msgs->outMessages;
1868
1869            /* build the data message */
1870            tr_bencInitDict( &tmp, 3 );
1871            tr_bencDictAddInt( &tmp, "msg_type", METADATA_MSG_TYPE_DATA );
1872            tr_bencDictAddInt( &tmp, "piece", piece );
1873            tr_bencDictAddInt( &tmp, "total_size", msgs->torrent->infoDictLength );
1874            payload = tr_bencToStr( &tmp, TR_FMT_BENC, &payloadLen );
1875            tr_bencFree( &tmp );
1876
1877            /* write it out as a LTEP message to our outMessages buffer */
1878            evbuffer_add_uint32( out, 2 * sizeof( uint8_t ) + payloadLen + dataLen );
1879            evbuffer_add_uint8 ( out, BT_LTEP );
1880            evbuffer_add_uint8 ( out, msgs->ut_metadata_id );
1881            evbuffer_add       ( out, payload, payloadLen );
1882            evbuffer_add       ( out, data, dataLen );
1883            pokeBatchPeriod( msgs, HIGH_PRIORITY_INTERVAL_SECS );
1884            dbgOutMessageLen( msgs );
1885
1886            tr_free( payload );
1887            tr_free( data );
1888
1889            ok = TRUE;
1890        }
1891
1892        if( !ok ) /* send a rejection message */
1893        {
1894            tr_benc tmp;
1895            int payloadLen;
1896            char * payload;
1897            struct evbuffer * out = msgs->outMessages;
1898
1899            /* build the rejection message */
1900            tr_bencInitDict( &tmp, 2 );
1901            tr_bencDictAddInt( &tmp, "msg_type", METADATA_MSG_TYPE_REJECT );
1902            tr_bencDictAddInt( &tmp, "piece", piece );
1903            payload = tr_bencToStr( &tmp, TR_FMT_BENC, &payloadLen );
1904            tr_bencFree( &tmp );
1905
1906            /* write it out as a LTEP message to our outMessages buffer */
1907            evbuffer_add_uint32( out, 2 * sizeof( uint8_t ) + payloadLen );
1908            evbuffer_add_uint8 ( out, BT_LTEP );
1909            evbuffer_add_uint8 ( out, msgs->ut_metadata_id );
1910            evbuffer_add       ( out, payload, payloadLen );
1911            pokeBatchPeriod( msgs, HIGH_PRIORITY_INTERVAL_SECS );
1912            dbgOutMessageLen( msgs );
1913
1914            tr_free( payload );
1915        }
1916    }
1917
1918    /**
1919    ***  Data Blocks
1920    **/
1921
1922    if( ( tr_peerIoGetWriteBufferSpace( msgs->peer->io, now ) >= msgs->torrent->blockSize )
1923        && popNextRequest( msgs, &req ) )
1924    {
1925        --msgs->prefetchCount;
1926
1927        if( requestIsValid( msgs, &req )
1928            && tr_cpPieceIsComplete( &msgs->torrent->completion, req.index ) )
1929        {
1930            int err;
1931            const uint32_t msglen = 4 + 1 + 4 + 4 + req.length;
1932            struct evbuffer * out;
1933            struct evbuffer_iovec iovec[1];
1934
1935            out = evbuffer_new( );
1936            evbuffer_expand( out, msglen );
1937
1938            evbuffer_add_uint32( out, sizeof( uint8_t ) + 2 * sizeof( uint32_t ) + req.length );
1939            evbuffer_add_uint8 ( out, BT_PIECE );
1940            evbuffer_add_uint32( out, req.index );
1941            evbuffer_add_uint32( out, req.offset );
1942
1943            evbuffer_reserve_space( out, req.length, iovec, 1 );
1944            err = tr_cacheReadBlock( getSession(msgs)->cache, msgs->torrent, req.index, req.offset, req.length, iovec[0].iov_base );
1945            iovec[0].iov_len = req.length;
1946            evbuffer_commit_space( out, iovec, 1 );
1947
1948            /* check the piece if it needs checking... */
1949            if( !err && tr_torrentPieceNeedsCheck( msgs->torrent, req.index ) )
1950                if(( err = !tr_torrentCheckPiece( msgs->torrent, req.index )))
1951                    tr_torrentSetLocalError( msgs->torrent, _( "Please Verify Local Data! Piece #%zu is corrupt." ), (size_t)req.index );
1952
1953            if( err )
1954            {
1955                if( fext )
1956                    protocolSendReject( msgs, &req );
1957            }
1958            else
1959            {
1960                const size_t n = evbuffer_get_length( out );
1961                dbgmsg( msgs, "sending block %u:%u->%u", req.index, req.offset, req.length );
1962                assert( n == msglen );
1963                tr_peerIoWriteBuf( msgs->peer->io, out, TRUE );
1964                bytesWritten += n;
1965                msgs->clientSentAnythingAt = now;
1966                tr_historyAdd( msgs->peer->blocksSentToPeer, tr_time( ), 1 );
1967            }
1968
1969            evbuffer_free( out );
1970
1971            if( err )
1972            {
1973                bytesWritten = 0;
1974                msgs = NULL;
1975            }
1976        }
1977        else if( fext ) /* peer needs a reject message */
1978        {
1979            protocolSendReject( msgs, &req );
1980        }
1981
1982        if( msgs != NULL )
1983            prefetchPieces( msgs );
1984    }
1985
1986    /**
1987    ***  Keepalive
1988    **/
1989
1990    if( ( msgs != NULL )
1991        && ( msgs->clientSentAnythingAt != 0 )
1992        && ( ( now - msgs->clientSentAnythingAt ) > KEEPALIVE_INTERVAL_SECS ) )
1993    {
1994        dbgmsg( msgs, "sending a keepalive message" );
1995        evbuffer_add_uint32( msgs->outMessages, 0 );
1996        pokeBatchPeriod( msgs, IMMEDIATE_PRIORITY_INTERVAL_SECS );
1997    }
1998
1999    return bytesWritten;
2000}
2001
2002static int
2003peerPulse( void * vmsgs )
2004{
2005    tr_peermsgs * msgs = vmsgs;
2006    const time_t  now = tr_time( );
2007
2008    if ( tr_isPeerIo( msgs->peer->io ) ) {
2009        updateDesiredRequestCount( msgs );
2010        updateBlockRequests( msgs );
2011        updateMetadataRequests( msgs, now );
2012    }
2013
2014    for( ;; )
2015        if( fillOutputBuffer( msgs, now ) < 1 )
2016            break;
2017
2018    return TRUE; /* loop forever */
2019}
2020
2021void
2022tr_peerMsgsPulse( tr_peermsgs * msgs )
2023{
2024    if( msgs != NULL )
2025        peerPulse( msgs );
2026}
2027
2028static void
2029gotError( tr_peerIo  * io UNUSED,
2030          short        what,
2031          void       * vmsgs )
2032{
2033    if( what & BEV_EVENT_TIMEOUT )
2034        dbgmsg( vmsgs, "libevent got a timeout, what=%hd", what );
2035    if( what & ( BEV_EVENT_EOF | BEV_EVENT_ERROR ) )
2036        dbgmsg( vmsgs, "libevent got an error! what=%hd, errno=%d (%s)",
2037               what, errno, tr_strerror( errno ) );
2038    fireError( vmsgs, ENOTCONN );
2039}
2040
2041static void
2042sendBitfield( tr_peermsgs * msgs )
2043{
2044    struct evbuffer * out = msgs->outMessages;
2045    tr_bitfield *     field;
2046    tr_piece_index_t  lazyPieces[LAZY_PIECE_COUNT];
2047    size_t            i;
2048    size_t            lazyCount = 0;
2049
2050    field = tr_bitfieldDup( tr_cpPieceBitfield( &msgs->torrent->completion ) );
2051
2052    if( tr_sessionIsLazyBitfieldEnabled( getSession( msgs ) ) )
2053    {
2054        /** Lazy bitfields aren't a high priority or secure, so I'm opting for
2055            speed over a truly random sample -- let's limit the pool size to
2056            the first 1000 pieces so large torrents don't bog things down */
2057        size_t poolSize;
2058        const size_t maxPoolSize = MIN( msgs->torrent->info.pieceCount, 1000 );
2059        tr_piece_index_t * pool = tr_new( tr_piece_index_t, maxPoolSize );
2060
2061        /* build the pool */
2062        for( i=poolSize=0; i<maxPoolSize; ++i )
2063            if( tr_bitfieldHas( field, i ) )
2064                pool[poolSize++] = i;
2065
2066        /* pull random piece indices from the pool */
2067        while( ( poolSize > 0 ) && ( lazyCount < LAZY_PIECE_COUNT ) )
2068        {
2069            const int pos = tr_cryptoWeakRandInt( poolSize );
2070            const tr_piece_index_t piece = pool[pos];
2071            tr_bitfieldRem( field, piece );
2072            lazyPieces[lazyCount++] = piece;
2073            pool[pos] = pool[--poolSize];
2074        }
2075
2076        /* cleanup */
2077        tr_free( pool );
2078    }
2079
2080    evbuffer_add_uint32( out, sizeof( uint8_t ) + field->byteCount );
2081    evbuffer_add_uint8 ( out, BT_BITFIELD );
2082    evbuffer_add       ( out, field->bits, field->byteCount );
2083    dbgmsg( msgs, "sending bitfield... outMessage size is now %zu",
2084            evbuffer_get_length( out ) );
2085    pokeBatchPeriod( msgs, IMMEDIATE_PRIORITY_INTERVAL_SECS );
2086
2087    for( i = 0; i < lazyCount; ++i )
2088        protocolSendHave( msgs, lazyPieces[i] );
2089
2090    tr_bitfieldFree( field );
2091}
2092
2093static void
2094tellPeerWhatWeHave( tr_peermsgs * msgs )
2095{
2096    const tr_bool fext = tr_peerIoSupportsFEXT( msgs->peer->io );
2097
2098    if( fext && ( tr_cpGetStatus( &msgs->torrent->completion ) == TR_SEED ) )
2099    {
2100        protocolSendHaveAll( msgs );
2101    }
2102    else if( fext && ( tr_cpHaveValid( &msgs->torrent->completion ) == 0 ) )
2103    {
2104        protocolSendHaveNone( msgs );
2105    }
2106    else
2107    {
2108        sendBitfield( msgs );
2109    }
2110}
2111
2112/**
2113***
2114**/
2115
2116/* some peers give us error messages if we send
2117   more than this many peers in a single pex message
2118   http://wiki.theory.org/BitTorrentPeerExchangeConventions */
2119#define MAX_PEX_ADDED 50
2120#define MAX_PEX_DROPPED 50
2121
2122typedef struct
2123{
2124    tr_pex *  added;
2125    tr_pex *  dropped;
2126    tr_pex *  elements;
2127    int       addedCount;
2128    int       droppedCount;
2129    int       elementCount;
2130}
2131PexDiffs;
2132
2133static void
2134pexAddedCb( void * vpex,
2135            void * userData )
2136{
2137    PexDiffs * diffs = userData;
2138    tr_pex *   pex = vpex;
2139
2140    if( diffs->addedCount < MAX_PEX_ADDED )
2141    {
2142        diffs->added[diffs->addedCount++] = *pex;
2143        diffs->elements[diffs->elementCount++] = *pex;
2144    }
2145}
2146
2147static inline void
2148pexDroppedCb( void * vpex,
2149              void * userData )
2150{
2151    PexDiffs * diffs = userData;
2152    tr_pex *   pex = vpex;
2153
2154    if( diffs->droppedCount < MAX_PEX_DROPPED )
2155    {
2156        diffs->dropped[diffs->droppedCount++] = *pex;
2157    }
2158}
2159
2160static inline void
2161pexElementCb( void * vpex,
2162              void * userData )
2163{
2164    PexDiffs * diffs = userData;
2165    tr_pex * pex = vpex;
2166
2167    diffs->elements[diffs->elementCount++] = *pex;
2168}
2169
2170static void
2171sendPex( tr_peermsgs * msgs )
2172{
2173    if( msgs->peerSupportsPex && tr_torrentAllowsPex( msgs->torrent ) )
2174    {
2175        PexDiffs diffs;
2176        PexDiffs diffs6;
2177        tr_pex * newPex = NULL;
2178        tr_pex * newPex6 = NULL;
2179        const int newCount = tr_peerMgrGetPeers( msgs->torrent, &newPex, TR_AF_INET, TR_PEERS_CONNECTED, MAX_PEX_PEER_COUNT );
2180        const int newCount6 = tr_peerMgrGetPeers( msgs->torrent, &newPex6, TR_AF_INET6, TR_PEERS_CONNECTED, MAX_PEX_PEER_COUNT );
2181
2182        /* build the diffs */
2183        diffs.added = tr_new( tr_pex, newCount );
2184        diffs.addedCount = 0;
2185        diffs.dropped = tr_new( tr_pex, msgs->pexCount );
2186        diffs.droppedCount = 0;
2187        diffs.elements = tr_new( tr_pex, newCount + msgs->pexCount );
2188        diffs.elementCount = 0;
2189        tr_set_compare( msgs->pex, msgs->pexCount,
2190                        newPex, newCount,
2191                        tr_pexCompare, sizeof( tr_pex ),
2192                        pexDroppedCb, pexAddedCb, pexElementCb, &diffs );
2193        diffs6.added = tr_new( tr_pex, newCount6 );
2194        diffs6.addedCount = 0;
2195        diffs6.dropped = tr_new( tr_pex, msgs->pexCount6 );
2196        diffs6.droppedCount = 0;
2197        diffs6.elements = tr_new( tr_pex, newCount6 + msgs->pexCount6 );
2198        diffs6.elementCount = 0;
2199        tr_set_compare( msgs->pex6, msgs->pexCount6,
2200                        newPex6, newCount6,
2201                        tr_pexCompare, sizeof( tr_pex ),
2202                        pexDroppedCb, pexAddedCb, pexElementCb, &diffs6 );
2203        dbgmsg(
2204            msgs,
2205            "pex: old peer count %d+%d, new peer count %d+%d, "
2206            "added %d+%d, removed %d+%d",
2207            msgs->pexCount, msgs->pexCount6, newCount, newCount6,
2208            diffs.addedCount, diffs6.addedCount,
2209            diffs.droppedCount, diffs6.droppedCount );
2210
2211        if( !diffs.addedCount && !diffs.droppedCount && !diffs6.addedCount &&
2212            !diffs6.droppedCount )
2213        {
2214            tr_free( diffs.elements );
2215            tr_free( diffs6.elements );
2216        }
2217        else
2218        {
2219            int  i;
2220            tr_benc val;
2221            char * benc;
2222            int bencLen;
2223            uint8_t * tmp, *walk;
2224            struct evbuffer * out = msgs->outMessages;
2225
2226            /* update peer */
2227            tr_free( msgs->pex );
2228            msgs->pex = diffs.elements;
2229            msgs->pexCount = diffs.elementCount;
2230            tr_free( msgs->pex6 );
2231            msgs->pex6 = diffs6.elements;
2232            msgs->pexCount6 = diffs6.elementCount;
2233
2234            /* build the pex payload */
2235            tr_bencInitDict( &val, 3 ); /* ipv6 support: left as 3:
2236                                         * speed vs. likelihood? */
2237
2238            if( diffs.addedCount > 0)
2239            {
2240                /* "added" */
2241                tmp = walk = tr_new( uint8_t, diffs.addedCount * 6 );
2242                for( i = 0; i < diffs.addedCount; ++i ) {
2243                    memcpy( walk, &diffs.added[i].addr.addr, 4 ); walk += 4;
2244                    memcpy( walk, &diffs.added[i].port, 2 ); walk += 2;
2245                }
2246                assert( ( walk - tmp ) == diffs.addedCount * 6 );
2247                tr_bencDictAddRaw( &val, "added", tmp, walk - tmp );
2248                tr_free( tmp );
2249
2250                /* "added.f" */
2251                tmp = walk = tr_new( uint8_t, diffs.addedCount );
2252                for( i = 0; i < diffs.addedCount; ++i )
2253                    *walk++ = diffs.added[i].flags;
2254                assert( ( walk - tmp ) == diffs.addedCount );
2255                tr_bencDictAddRaw( &val, "added.f", tmp, walk - tmp );
2256                tr_free( tmp );
2257            }
2258
2259            if( diffs.droppedCount > 0 )
2260            {
2261                /* "dropped" */
2262                tmp = walk = tr_new( uint8_t, diffs.droppedCount * 6 );
2263                for( i = 0; i < diffs.droppedCount; ++i ) {
2264                    memcpy( walk, &diffs.dropped[i].addr.addr, 4 ); walk += 4;
2265                    memcpy( walk, &diffs.dropped[i].port, 2 ); walk += 2;
2266                }
2267                assert( ( walk - tmp ) == diffs.droppedCount * 6 );
2268                tr_bencDictAddRaw( &val, "dropped", tmp, walk - tmp );
2269                tr_free( tmp );
2270            }
2271
2272            if( diffs6.addedCount > 0 )
2273            {
2274                /* "added6" */
2275                tmp = walk = tr_new( uint8_t, diffs6.addedCount * 18 );
2276                for( i = 0; i < diffs6.addedCount; ++i ) {
2277                    memcpy( walk, &diffs6.added[i].addr.addr.addr6.s6_addr, 16 );
2278                    walk += 16;
2279                    memcpy( walk, &diffs6.added[i].port, 2 );
2280                    walk += 2;
2281                }
2282                assert( ( walk - tmp ) == diffs6.addedCount * 18 );
2283                tr_bencDictAddRaw( &val, "added6", tmp, walk - tmp );
2284                tr_free( tmp );
2285
2286                /* "added6.f" */
2287                tmp = walk = tr_new( uint8_t, diffs6.addedCount );
2288                for( i = 0; i < diffs6.addedCount; ++i )
2289                    *walk++ = diffs6.added[i].flags;
2290                assert( ( walk - tmp ) == diffs6.addedCount );
2291                tr_bencDictAddRaw( &val, "added6.f", tmp, walk - tmp );
2292                tr_free( tmp );
2293            }
2294
2295            if( diffs6.droppedCount > 0 )
2296            {
2297                /* "dropped6" */
2298                tmp = walk = tr_new( uint8_t, diffs6.droppedCount * 18 );
2299                for( i = 0; i < diffs6.droppedCount; ++i ) {
2300                    memcpy( walk, &diffs6.dropped[i].addr.addr.addr6.s6_addr, 16 );
2301                    walk += 16;
2302                    memcpy( walk, &diffs6.dropped[i].port, 2 );
2303                    walk += 2;
2304                }
2305                assert( ( walk - tmp ) == diffs6.droppedCount * 18);
2306                tr_bencDictAddRaw( &val, "dropped6", tmp, walk - tmp );
2307                tr_free( tmp );
2308            }
2309
2310            /* write the pex message */
2311            benc = tr_bencToStr( &val, TR_FMT_BENC, &bencLen );
2312            evbuffer_add_uint32( out, 2 * sizeof( uint8_t ) + bencLen );
2313            evbuffer_add_uint8 ( out, BT_LTEP );
2314            evbuffer_add_uint8 ( out, msgs->ut_pex_id );
2315            evbuffer_add       ( out, benc, bencLen );
2316            pokeBatchPeriod( msgs, HIGH_PRIORITY_INTERVAL_SECS );
2317            dbgmsg( msgs, "sending a pex message; outMessage size is now %zu", evbuffer_get_length( out ) );
2318            dbgOutMessageLen( msgs );
2319
2320            tr_free( benc );
2321            tr_bencFree( &val );
2322        }
2323
2324        /* cleanup */
2325        tr_free( diffs.added );
2326        tr_free( diffs.dropped );
2327        tr_free( newPex );
2328        tr_free( diffs6.added );
2329        tr_free( diffs6.dropped );
2330        tr_free( newPex6 );
2331
2332        /*msgs->clientSentPexAt = tr_time( );*/
2333    }
2334}
2335
2336static void
2337pexPulse( int foo UNUSED, short bar UNUSED, void * vmsgs )
2338{
2339    struct tr_peermsgs * msgs = vmsgs;
2340
2341    sendPex( msgs );
2342
2343    tr_timerAdd( msgs->pexTimer, PEX_INTERVAL_SECS, 0 );
2344}
2345
2346/**
2347***
2348**/
2349
2350tr_peermsgs*
2351tr_peerMsgsNew( struct tr_torrent    * torrent,
2352                struct tr_peer       * peer,
2353                tr_peer_callback     * callback,
2354                void                 * callbackData )
2355{
2356    tr_peermsgs * m;
2357
2358    assert( peer );
2359    assert( peer->io );
2360
2361    m = tr_new0( tr_peermsgs, 1 );
2362    m->callback = callback;
2363    m->callbackData = callbackData;
2364    m->peer = peer;
2365    m->torrent = torrent;
2366    m->peer->clientIsChoked = 1;
2367    m->peer->peerIsChoked = 1;
2368    m->peer->clientIsInterested = 0;
2369    m->peer->peerIsInterested = 0;
2370    m->state = AWAITING_BT_LENGTH;
2371    m->outMessages = evbuffer_new( );
2372    m->outMessagesBatchedAt = 0;
2373    m->outMessagesBatchPeriod = LOW_PRIORITY_INTERVAL_SECS;
2374    m->incoming.block = evbuffer_new( );
2375    m->pexTimer = evtimer_new( torrent->session->event_base, pexPulse, m );
2376    peer->msgs = m;
2377    tr_timerAdd( m->pexTimer, PEX_INTERVAL_SECS, 0 );
2378
2379    if( tr_peerIoSupportsUTP( peer->io ) ) {
2380        const tr_address * addr = tr_peerIoGetAddress( peer->io, NULL );
2381        tr_peerMgrSetUtpSupported( torrent, addr );
2382    }
2383
2384    if( tr_peerIoSupportsLTEP( peer->io ) )
2385        sendLtepHandshake( m );
2386
2387    tellPeerWhatWeHave( m );
2388
2389    if( tr_dhtEnabled( torrent->session ) && tr_peerIoSupportsDHT( peer->io ))
2390    {
2391        /* Only send PORT over IPv6 when the IPv6 DHT is running (BEP-32). */
2392        const struct tr_address *addr = tr_peerIoGetAddress( peer->io, NULL );
2393        if( addr->type == TR_AF_INET || tr_globalIPv6() ) {
2394            protocolSendPort( m, tr_dhtPort( torrent->session ) );
2395        }
2396    }
2397
2398    tr_peerIoSetIOFuncs( m->peer->io, canRead, didWrite, gotError, m );
2399    updateDesiredRequestCount( m );
2400
2401    return m;
2402}
2403
2404void
2405tr_peerMsgsFree( tr_peermsgs* msgs )
2406{
2407    if( msgs )
2408    {
2409        event_free( msgs->pexTimer );
2410
2411        evbuffer_free( msgs->incoming.block );
2412        evbuffer_free( msgs->outMessages );
2413        tr_free( msgs->pex6 );
2414        tr_free( msgs->pex );
2415
2416        memset( msgs, ~0, sizeof( tr_peermsgs ) );
2417        tr_free( msgs );
2418    }
2419}
Note: See TracBrowser for help on using the repository browser.