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

Last change on this file since 9593 was 9593, checked in by charles, 11 years ago

(trunk libT) #2610 "avoid unnecessary calls to time(NULL)"

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