/* * This file Copyright (C) 2007-2009 Charles Kerr * * This file is licensed by the GPL version 2. Works owned by the * Transmission project are granted a special exemption to clause 2(b) * so that the bulk of its code can remain under the MIT license. * This exemption does not extend to derived works not owned by * the Transmission project. * * $Id$ */ #include #include #include /* INT_MAX */ #include #include #include #include #include "transmission.h" #include "session.h" #include "bencode.h" #include "completion.h" #include "crypto.h" #include "inout.h" #ifdef WIN32 #include "net.h" /* for ECONN */ #endif #include "peer-io.h" #include "peer-mgr.h" #include "peer-msgs.h" #include "platform.h" /* MAX_STACK_ARRAY_SIZE */ #include "ratecontrol.h" #include "request-list.h" #include "stats.h" #include "torrent.h" #include "tr-dht.h" #include "utils.h" #include "version.h" /** *** **/ enum { BT_CHOKE = 0, BT_UNCHOKE = 1, BT_INTERESTED = 2, BT_NOT_INTERESTED = 3, BT_HAVE = 4, BT_BITFIELD = 5, BT_REQUEST = 6, BT_PIECE = 7, BT_CANCEL = 8, BT_PORT = 9, BT_FEXT_SUGGEST = 13, BT_FEXT_HAVE_ALL = 14, BT_FEXT_HAVE_NONE = 15, BT_FEXT_REJECT = 16, BT_FEXT_ALLOWED_FAST = 17, BT_LTEP = 20, LTEP_HANDSHAKE = 0, TR_LTEP_PEX = 1, MIN_CHOKE_PERIOD_SEC = ( 10 ), /* idle seconds before we send a keepalive */ KEEPALIVE_INTERVAL_SECS = 100, PEX_INTERVAL_SECS = 90, /* sec between sendPex() calls */ MAX_BLOCK_SIZE = ( 1024 * 16 ), /* how long an unsent request can stay queued before it's returned back to the peer-mgr's pool of requests */ QUEUED_REQUEST_TTL_SECS = 20, /* how long a sent request can stay queued before it's returned back to the peer-mgr's pool of requests */ SENT_REQUEST_TTL_SECS = 240, /* used in lowering the outMessages queue period */ IMMEDIATE_PRIORITY_INTERVAL_SECS = 0, HIGH_PRIORITY_INTERVAL_SECS = 2, LOW_PRIORITY_INTERVAL_SECS = 20, /* number of pieces to remove from the bitfield when * lazy bitfields are turned on */ LAZY_PIECE_COUNT = 26, /* number of pieces we'll allow in our fast set */ MAX_FAST_SET_SIZE = 3 }; enum { AWAITING_BT_LENGTH, AWAITING_BT_ID, AWAITING_BT_MESSAGE, AWAITING_BT_PIECE }; /** *** **/ /* this is raw, unchanged data from the peer regarding * the current message that it's sending us. */ struct tr_incoming { uint8_t id; uint32_t length; /* includes the +1 for id length */ struct peer_request blockReq; /* metadata for incoming blocks */ struct evbuffer * block; /* piece data for incoming blocks */ }; /** * Low-level communication state information about a connected peer. * * This structure remembers the low-level protocol states that we're * in with this peer, such as active requests, pex messages, and so on. * Its fields are all private to peer-msgs.c. * * Data not directly involved with sending & receiving messages is * stored in tr_peer, where it can be accessed by both peermsgs and * the peer manager. * * @see struct peer_atom * @see tr_peer */ struct tr_peermsgs { tr_bool peerSupportsPex; tr_bool clientSentLtepHandshake; tr_bool peerSentLtepHandshake; /*tr_bool haveFastSet;*/ /* how long the outMessages batch should be allowed to grow before * it's flushed -- some messages (like requests >:) should be sent * very quickly; others aren't as urgent. */ int8_t outMessagesBatchPeriod; uint8_t state; uint8_t ut_pex_id; uint16_t pexCount; uint16_t pexCount6; uint16_t maxActiveRequests; #if 0 size_t fastsetSize; tr_piece_index_t fastset[MAX_FAST_SET_SIZE]; #endif tr_peer * peer; tr_torrent * torrent; tr_publisher publisher; struct evbuffer * outMessages; /* all the non-piece messages */ struct request_list peerAskedFor; struct request_list clientAskedFor; struct request_list clientWillAskFor; tr_pex * pex; tr_pex * pex6; /*time_t clientSentPexAt;*/ time_t clientSentAnythingAt; /* when we started batching the outMessages */ time_t outMessagesBatchedAt; struct tr_incoming incoming; /* if the peer supports the Extension Protocol in BEP 10 and supplied a reqq argument, it's stored here. otherwise the value is zero and should be ignored. */ int64_t reqq; struct event pexTimer; }; static TR_INLINE tr_session* getSession( struct tr_peermsgs * msgs ) { return msgs->torrent->session; } /** *** **/ static void myDebug( const char * file, int line, const struct tr_peermsgs * msgs, const char * fmt, ... ) { FILE * fp = tr_getLog( ); if( fp ) { va_list args; char timestr[64]; struct evbuffer * buf = evbuffer_new( ); char * base = tr_basename( file ); evbuffer_add_printf( buf, "[%s] %s - %s [%s]: ", tr_getLogTimeStr( timestr, sizeof( timestr ) ), msgs->torrent->info.name, tr_peerIoGetAddrStr( msgs->peer->io ), msgs->peer->client ); va_start( args, fmt ); evbuffer_add_vprintf( buf, fmt, args ); va_end( args ); evbuffer_add_printf( buf, " (%s:%d)\n", base, line ); /* FIXME(libevent2) tr_getLog() should return an fd, then use evbuffer_write() here */ fwrite( EVBUFFER_DATA( buf ), 1, EVBUFFER_LENGTH( buf ), fp ); tr_free( base ); evbuffer_free( buf ); } } #define dbgmsg( msgs, ... ) \ do { \ if( tr_deepLoggingIsActive( ) ) \ myDebug( __FILE__, __LINE__, msgs, __VA_ARGS__ ); \ } while( 0 ) /** *** **/ static void pokeBatchPeriod( tr_peermsgs * msgs, int interval ) { if( msgs->outMessagesBatchPeriod > interval ) { msgs->outMessagesBatchPeriod = interval; dbgmsg( msgs, "lowering batch interval to %d seconds", interval ); } } static TR_INLINE void dbgOutMessageLen( tr_peermsgs * msgs ) { dbgmsg( msgs, "outMessage size is now %zu", EVBUFFER_LENGTH( msgs->outMessages ) ); } static void protocolSendReject( tr_peermsgs * msgs, const struct peer_request * req ) { tr_peerIo * io = msgs->peer->io; struct evbuffer * out = msgs->outMessages; assert( tr_peerIoSupportsFEXT( msgs->peer->io ) ); tr_peerIoWriteUint32( io, out, sizeof( uint8_t ) + 3 * sizeof( uint32_t ) ); tr_peerIoWriteUint8 ( io, out, BT_FEXT_REJECT ); tr_peerIoWriteUint32( io, out, req->index ); tr_peerIoWriteUint32( io, out, req->offset ); tr_peerIoWriteUint32( io, out, req->length ); dbgmsg( msgs, "rejecting %u:%u->%u...", req->index, req->offset, req->length ); dbgOutMessageLen( msgs ); } static void protocolSendRequest( tr_peermsgs * msgs, const struct peer_request * req ) { tr_peerIo * io = msgs->peer->io; struct evbuffer * out = msgs->outMessages; tr_peerIoWriteUint32( io, out, sizeof( uint8_t ) + 3 * sizeof( uint32_t ) ); tr_peerIoWriteUint8 ( io, out, BT_REQUEST ); tr_peerIoWriteUint32( io, out, req->index ); tr_peerIoWriteUint32( io, out, req->offset ); tr_peerIoWriteUint32( io, out, req->length ); dbgmsg( msgs, "requesting %u:%u->%u...", req->index, req->offset, req->length ); dbgOutMessageLen( msgs ); pokeBatchPeriod( msgs, IMMEDIATE_PRIORITY_INTERVAL_SECS ); } static void protocolSendCancel( tr_peermsgs * msgs, const struct peer_request * req ) { tr_peerIo * io = msgs->peer->io; struct evbuffer * out = msgs->outMessages; tr_peerIoWriteUint32( io, out, sizeof( uint8_t ) + 3 * sizeof( uint32_t ) ); tr_peerIoWriteUint8 ( io, out, BT_CANCEL ); tr_peerIoWriteUint32( io, out, req->index ); tr_peerIoWriteUint32( io, out, req->offset ); tr_peerIoWriteUint32( io, out, req->length ); dbgmsg( msgs, "cancelling %u:%u->%u...", req->index, req->offset, req->length ); dbgOutMessageLen( msgs ); pokeBatchPeriod( msgs, IMMEDIATE_PRIORITY_INTERVAL_SECS ); } static void protocolSendPort(tr_peermsgs *msgs, uint16_t port) { tr_peerIo * io = msgs->peer->io; struct evbuffer * out = msgs->outMessages; dbgmsg( msgs, "sending Port %u", port); tr_peerIoWriteUint32( io, out, 3 ); tr_peerIoWriteUint8 ( io, out, BT_PORT ); tr_peerIoWriteUint16( io, out, port); } static void protocolSendHave( tr_peermsgs * msgs, uint32_t index ) { tr_peerIo * io = msgs->peer->io; struct evbuffer * out = msgs->outMessages; tr_peerIoWriteUint32( io, out, sizeof(uint8_t) + sizeof(uint32_t) ); tr_peerIoWriteUint8 ( io, out, BT_HAVE ); tr_peerIoWriteUint32( io, out, index ); dbgmsg( msgs, "sending Have %u", index ); dbgOutMessageLen( msgs ); pokeBatchPeriod( msgs, LOW_PRIORITY_INTERVAL_SECS ); } #if 0 static void protocolSendAllowedFast( tr_peermsgs * msgs, uint32_t pieceIndex ) { tr_peerIo * io = msgs->peer->io; struct evbuffer * out = msgs->outMessages; assert( tr_peerIoSupportsFEXT( msgs->peer->io ) ); tr_peerIoWriteUint32( io, out, sizeof(uint8_t) + sizeof(uint32_t) ); tr_peerIoWriteUint8 ( io, out, BT_FEXT_ALLOWED_FAST ); tr_peerIoWriteUint32( io, out, pieceIndex ); dbgmsg( msgs, "sending Allowed Fast %u...", pieceIndex ); dbgOutMessageLen( msgs ); } #endif static void protocolSendChoke( tr_peermsgs * msgs, int choke ) { tr_peerIo * io = msgs->peer->io; struct evbuffer * out = msgs->outMessages; tr_peerIoWriteUint32( io, out, sizeof( uint8_t ) ); tr_peerIoWriteUint8 ( io, out, choke ? BT_CHOKE : BT_UNCHOKE ); dbgmsg( msgs, "sending %s...", choke ? "Choke" : "Unchoke" ); dbgOutMessageLen( msgs ); pokeBatchPeriod( msgs, IMMEDIATE_PRIORITY_INTERVAL_SECS ); } static void protocolSendHaveAll( tr_peermsgs * msgs ) { tr_peerIo * io = msgs->peer->io; struct evbuffer * out = msgs->outMessages; assert( tr_peerIoSupportsFEXT( msgs->peer->io ) ); tr_peerIoWriteUint32( io, out, sizeof( uint8_t ) ); tr_peerIoWriteUint8 ( io, out, BT_FEXT_HAVE_ALL ); dbgmsg( msgs, "sending HAVE_ALL..." ); dbgOutMessageLen( msgs ); pokeBatchPeriod( msgs, IMMEDIATE_PRIORITY_INTERVAL_SECS ); } static void protocolSendHaveNone( tr_peermsgs * msgs ) { tr_peerIo * io = msgs->peer->io; struct evbuffer * out = msgs->outMessages; assert( tr_peerIoSupportsFEXT( msgs->peer->io ) ); tr_peerIoWriteUint32( io, out, sizeof( uint8_t ) ); tr_peerIoWriteUint8 ( io, out, BT_FEXT_HAVE_NONE ); dbgmsg( msgs, "sending HAVE_NONE..." ); dbgOutMessageLen( msgs ); pokeBatchPeriod( msgs, IMMEDIATE_PRIORITY_INTERVAL_SECS ); } /** *** EVENTS **/ static const tr_peer_event blankEvent = { 0, 0, 0, 0, 0.0f, 0, 0, 0 }; static void publish( tr_peermsgs * msgs, tr_peer_event * e ) { assert( msgs->peer ); assert( msgs->peer->msgs == msgs ); tr_publisherPublish( &msgs->publisher, msgs->peer, e ); } static void fireError( tr_peermsgs * msgs, int err ) { tr_peer_event e = blankEvent; e.eventType = TR_PEER_ERROR; e.err = err; publish( msgs, &e ); } static void fireUploadOnly( tr_peermsgs * msgs, tr_bool uploadOnly ) { tr_peer_event e = blankEvent; e.eventType = TR_PEER_UPLOAD_ONLY; e.uploadOnly = uploadOnly; publish( msgs, &e ); } static void fireNeedReq( tr_peermsgs * msgs ) { tr_peer_event e = blankEvent; e.eventType = TR_PEER_NEED_REQ; publish( msgs, &e ); } static void firePeerProgress( tr_peermsgs * msgs ) { tr_peer_event e = blankEvent; e.eventType = TR_PEER_PEER_PROGRESS; e.progress = msgs->peer->progress; publish( msgs, &e ); } static void fireGotBlock( tr_peermsgs * msgs, const struct peer_request * req ) { tr_peer_event e = blankEvent; e.eventType = TR_PEER_CLIENT_GOT_BLOCK; e.pieceIndex = req->index; e.offset = req->offset; e.length = req->length; publish( msgs, &e ); } static void fireClientGotData( tr_peermsgs * msgs, uint32_t length, int wasPieceData ) { tr_peer_event e = blankEvent; e.length = length; e.eventType = TR_PEER_CLIENT_GOT_DATA; e.wasPieceData = wasPieceData; publish( msgs, &e ); } static void fireClientGotSuggest( tr_peermsgs * msgs, uint32_t pieceIndex ) { tr_peer_event e = blankEvent; e.eventType = TR_PEER_CLIENT_GOT_SUGGEST; e.pieceIndex = pieceIndex; publish( msgs, &e ); } static void fireClientGotAllowedFast( tr_peermsgs * msgs, uint32_t pieceIndex ) { tr_peer_event e = blankEvent; e.eventType = TR_PEER_CLIENT_GOT_ALLOWED_FAST; e.pieceIndex = pieceIndex; publish( msgs, &e ); } static void firePeerGotData( tr_peermsgs * msgs, uint32_t length, int wasPieceData ) { tr_peer_event e = blankEvent; e.length = length; e.eventType = TR_PEER_PEER_GOT_DATA; e.wasPieceData = wasPieceData; publish( msgs, &e ); } static void fireCancelledReq( tr_peermsgs * msgs, const struct peer_request * req ) { tr_peer_event e = blankEvent; e.eventType = TR_PEER_CANCEL; e.pieceIndex = req->index; e.offset = req->offset; e.length = req->length; publish( msgs, &e ); } /** *** ALLOWED FAST SET *** For explanation, see http://www.bittorrent.org/beps/bep_0006.html **/ size_t tr_generateAllowedSet( tr_piece_index_t * setmePieces, size_t desiredSetSize, size_t pieceCount, const uint8_t * infohash, const tr_address * addr ) { size_t setSize = 0; assert( setmePieces ); assert( desiredSetSize <= pieceCount ); assert( desiredSetSize ); assert( pieceCount ); assert( infohash ); assert( addr ); if( addr->type == TR_AF_INET ) { uint8_t w[SHA_DIGEST_LENGTH + 4], *walk=w; uint8_t x[SHA_DIGEST_LENGTH]; uint32_t ui32 = ntohl( htonl( addr->addr.addr4.s_addr ) & 0xffffff00 ); /* (1) */ memcpy( w, &ui32, sizeof( uint32_t ) ); walk += sizeof( uint32_t ); memcpy( walk, infohash, SHA_DIGEST_LENGTH ); /* (2) */ walk += SHA_DIGEST_LENGTH; tr_sha1( x, w, walk-w, NULL ); /* (3) */ assert( sizeof( w ) == walk-w ); while( setSizepeer->io ); const int peerIsNeedy = msgs->peer->progress < 0.10; if( fext && peerIsNeedy && !msgs->haveFastSet ) { size_t i; const struct tr_address * addr = tr_peerIoGetAddress( msgs->peer->io, NULL ); const tr_info * inf = &msgs->torrent->info; const size_t numwant = MIN( MAX_FAST_SET_SIZE, inf->pieceCount ); /* build the fast set */ msgs->fastsetSize = tr_generateAllowedSet( msgs->fastset, numwant, inf->pieceCount, inf->hash, addr ); msgs->haveFastSet = 1; /* send it to the peer */ for( i=0; ifastsetSize; ++i ) protocolSendAllowedFast( msgs, msgs->fastset[i] ); } #endif } /** *** INTEREST **/ static tr_bool isPieceInteresting( const tr_peermsgs * msgs, tr_piece_index_t piece ) { const tr_torrent * torrent = msgs->torrent; return ( !torrent->info.pieces[piece].dnd ) /* we want it */ && ( !tr_cpPieceIsComplete( &torrent->completion, piece ) ) /* !have */ && ( tr_bitfieldHas( msgs->peer->have, piece ) ); /* peer has it */ } /* "interested" means we'll ask for piece data if they unchoke us */ static tr_bool isPeerInteresting( const tr_peermsgs * msgs ) { tr_piece_index_t i; const tr_torrent * torrent; const tr_bitfield * bitfield; const int clientIsSeed = tr_torrentIsSeed( msgs->torrent ); if( clientIsSeed ) return FALSE; if( !tr_torrentIsPieceTransferAllowed( msgs->torrent, TR_PEER_TO_CLIENT ) ) return FALSE; torrent = msgs->torrent; bitfield = tr_cpPieceBitfield( &torrent->completion ); if( !msgs->peer->have ) return TRUE; assert( bitfield->byteCount == msgs->peer->have->byteCount ); for( i = 0; i < torrent->info.pieceCount; ++i ) if( isPieceInteresting( msgs, i ) ) return TRUE; return FALSE; } static void sendInterest( tr_peermsgs * msgs, int weAreInterested ) { struct evbuffer * out = msgs->outMessages; assert( msgs ); assert( weAreInterested == 0 || weAreInterested == 1 ); msgs->peer->clientIsInterested = weAreInterested; dbgmsg( msgs, "Sending %s", weAreInterested ? "Interested" : "Not Interested" ); tr_peerIoWriteUint32( msgs->peer->io, out, sizeof( uint8_t ) ); tr_peerIoWriteUint8 ( msgs->peer->io, out, weAreInterested ? BT_INTERESTED : BT_NOT_INTERESTED ); pokeBatchPeriod( msgs, HIGH_PRIORITY_INTERVAL_SECS ); dbgOutMessageLen( msgs ); } static void updateInterest( tr_peermsgs * msgs ) { const int i = isPeerInteresting( msgs ); if( i != msgs->peer->clientIsInterested ) sendInterest( msgs, i ); if( i ) fireNeedReq( msgs ); } static tr_bool popNextRequest( tr_peermsgs * msgs, struct peer_request * setme ) { return reqListPop( &msgs->peerAskedFor, setme ); } static void cancelAllRequestsToClient( tr_peermsgs * msgs ) { struct peer_request req; const int mustSendCancel = tr_peerIoSupportsFEXT( msgs->peer->io ); while( popNextRequest( msgs, &req ) ) if( mustSendCancel ) protocolSendReject( msgs, &req ); } void tr_peerMsgsSetChoke( tr_peermsgs * msgs, int choke ) { const time_t now = time( NULL ); const time_t fibrillationTime = now - MIN_CHOKE_PERIOD_SEC; assert( msgs ); assert( msgs->peer ); assert( choke == 0 || choke == 1 ); if( msgs->peer->chokeChangedAt > fibrillationTime ) { dbgmsg( msgs, "Not changing choke to %d to avoid fibrillation", choke ); } else if( msgs->peer->peerIsChoked != choke ) { msgs->peer->peerIsChoked = choke; if( choke ) cancelAllRequestsToClient( msgs ); protocolSendChoke( msgs, choke ); msgs->peer->chokeChangedAt = now; } } /** *** **/ void tr_peerMsgsHave( tr_peermsgs * msgs, uint32_t index ) { protocolSendHave( msgs, index ); /* since we have more pieces now, we might not be interested in this peer */ updateInterest( msgs ); } /** *** **/ static tr_bool reqIsValid( const tr_peermsgs * peer, uint32_t index, uint32_t offset, uint32_t length ) { return tr_torrentReqIsValid( peer->torrent, index, offset, length ); } static tr_bool requestIsValid( const tr_peermsgs * msgs, const struct peer_request * req ) { return reqIsValid( msgs, req->index, req->offset, req->length ); } static void expireFromList( tr_peermsgs * msgs, struct request_list * list, const time_t oldestAllowed ) { size_t i; struct request_list tmp = REQUEST_LIST_INIT; /* since the fifo list is sorted by time, the oldest will be first */ if( !list->len || ( list->fifo[0].time_requested >= oldestAllowed ) ) return; /* if we found one too old, start pruning them */ reqListCopy( &tmp, list ); for( i=0; itime_requested >= oldestAllowed ) break; tr_peerMsgsCancel( msgs, req->index, req->offset, req->length ); } reqListClear( &tmp ); } static void expireOldRequests( tr_peermsgs * msgs, const time_t now ) { time_t oldestAllowed; const tr_bool fext = tr_peerIoSupportsFEXT( msgs->peer->io ); dbgmsg( msgs, "entering `expire old requests' block" ); /* cancel requests that have been queued for too long */ oldestAllowed = now - QUEUED_REQUEST_TTL_SECS; expireFromList( msgs, &msgs->clientWillAskFor, oldestAllowed ); /* if the peer doesn't support "Reject Request", * cancel requests that were sent too long ago. */ if( !fext ) { oldestAllowed = now - SENT_REQUEST_TTL_SECS; expireFromList( msgs, &msgs->clientAskedFor, oldestAllowed ); } dbgmsg( msgs, "leaving `expire old requests' block" ); } static void pumpRequestQueue( tr_peermsgs * msgs, const time_t now ) { const int max = msgs->maxActiveRequests; int sent = 0; int len = msgs->clientAskedFor.len; struct peer_request req; dbgmsg( msgs, "clientIsChoked %d, download allowed %d, len %d, max %d, msgs->clientWillAskFor.len %d", (int)msgs->peer->clientIsChoked, (int)tr_torrentIsPieceTransferAllowed( msgs->torrent, TR_PEER_TO_CLIENT ), len, max, msgs->clientWillAskFor.len ); if( msgs->peer->clientIsChoked ) return; if( !tr_torrentIsPieceTransferAllowed( msgs->torrent, TR_PEER_TO_CLIENT ) ) return; while( ( len < max ) && reqListPop( &msgs->clientWillAskFor, &req ) ) { const tr_block_index_t block = _tr_block( msgs->torrent, req.index, req.offset ); assert( requestIsValid( msgs, &req ) ); assert( tr_bitfieldHas( msgs->peer->have, req.index ) ); /* don't ask for it if we've already got it... this block may have * come in from a different peer after we cancelled a request for it */ if( !tr_cpBlockIsComplete( &msgs->torrent->completion, block ) ) { protocolSendRequest( msgs, &req ); req.time_requested = now; reqListAppend( &msgs->clientAskedFor, &req ); ++len; ++sent; } else dbgmsg( msgs, "not asking for it because we've already got it..." ); } if( sent ) dbgmsg( msgs, "pump sent %d requests, now have %d active and %d queued", sent, msgs->clientAskedFor.len, msgs->clientWillAskFor.len ); if( len < max ) fireNeedReq( msgs ); } static TR_INLINE tr_bool requestQueueIsFull( const tr_peermsgs * msgs ) { const int req_max = msgs->maxActiveRequests; return msgs->clientWillAskFor.len >= (size_t)req_max; } tr_addreq_t tr_peerMsgsAddRequest( tr_peermsgs * msgs, uint32_t index, uint32_t offset, uint32_t length ) { struct peer_request req; assert( msgs ); assert( msgs->torrent ); /** *** Reasons to decline the request **/ /* don't send requests to choked clients */ if( msgs->peer->clientIsChoked ) { dbgmsg( msgs, "declining request because they're choking us" ); return TR_ADDREQ_CLIENT_CHOKED; } /* peer's queue is full */ if( requestQueueIsFull( msgs ) ) { dbgmsg( msgs, "declining request because we're full" ); return TR_ADDREQ_FULL; } /* peer doesn't have this piece */ if( !tr_bitfieldHas( msgs->peer->have, index ) ) return TR_ADDREQ_MISSING; /* have we already asked for this piece? */ req.index = index; req.offset = offset; req.length = length; if( reqListHas( &msgs->clientAskedFor, &req ) ) { dbgmsg( msgs, "declining because it's a duplicate" ); return TR_ADDREQ_DUPLICATE; } if( reqListHas( &msgs->clientWillAskFor, &req ) ) { dbgmsg( msgs, "declining because it's a duplicate" ); return TR_ADDREQ_DUPLICATE; } /** *** Accept this request **/ dbgmsg( msgs, "adding req for %"PRIu32":%"PRIu32"->%"PRIu32" to our `will request' list", index, offset, length ); req.time_requested = time( NULL ); reqListAppend( &msgs->clientWillAskFor, &req ); return TR_ADDREQ_OK; } static void cancelAllRequestsToPeer( tr_peermsgs * msgs, tr_bool sendCancel ) { size_t i; struct request_list a = msgs->clientWillAskFor; struct request_list b = msgs->clientAskedFor; dbgmsg( msgs, "cancelling all requests to peer" ); msgs->clientAskedFor = REQUEST_LIST_INIT; msgs->clientWillAskFor = REQUEST_LIST_INIT; for( i=0; i 0 ); /* have we asked the peer for this piece? */ req.index = pieceIndex; req.offset = offset; req.length = length; /* if it's only in the queue and hasn't been sent yet, free it */ if( reqListRemove( &msgs->clientWillAskFor, &req ) ) { dbgmsg( msgs, "cancelling %"PRIu32":%"PRIu32"->%"PRIu32, pieceIndex, offset, length ); fireCancelledReq( msgs, &req ); } /* if it's already been sent, send a cancel message too */ if( reqListRemove( &msgs->clientAskedFor, &req ) ) { dbgmsg( msgs, "cancelling %"PRIu32":%"PRIu32"->%"PRIu32, pieceIndex, offset, length ); protocolSendCancel( msgs, &req ); fireCancelledReq( msgs, &req ); } } /** *** **/ static void sendLtepHandshake( tr_peermsgs * msgs ) { tr_benc val, *m; char * buf; int len; int pex; struct evbuffer * out = msgs->outMessages; if( msgs->clientSentLtepHandshake ) return; dbgmsg( msgs, "sending an ltep handshake" ); msgs->clientSentLtepHandshake = 1; /* decide if we want to advertise pex support */ if( !tr_torrentAllowsPex( msgs->torrent ) ) pex = 0; else if( msgs->peerSentLtepHandshake ) pex = msgs->peerSupportsPex ? 1 : 0; else pex = 1; tr_bencInitDict( &val, 5 ); tr_bencDictAddInt( &val, "e", getSession(msgs)->encryptionMode != TR_CLEAR_PREFERRED ); tr_bencDictAddInt( &val, "p", tr_sessionGetPeerPort( getSession(msgs) ) ); tr_bencDictAddInt( &val, "upload_only", tr_torrentIsSeed( msgs->torrent ) ); tr_bencDictAddStr( &val, "v", TR_NAME " " USERAGENT_PREFIX ); m = tr_bencDictAddDict( &val, "m", 1 ); if( pex ) tr_bencDictAddInt( m, "ut_pex", TR_LTEP_PEX ); buf = tr_bencToStr( &val, TR_FMT_BENC, &len ); tr_peerIoWriteUint32( msgs->peer->io, out, 2 * sizeof( uint8_t ) + len ); tr_peerIoWriteUint8 ( msgs->peer->io, out, BT_LTEP ); tr_peerIoWriteUint8 ( msgs->peer->io, out, LTEP_HANDSHAKE ); tr_peerIoWriteBytes ( msgs->peer->io, out, buf, len ); pokeBatchPeriod( msgs, IMMEDIATE_PRIORITY_INTERVAL_SECS ); dbgOutMessageLen( msgs ); /* cleanup */ tr_bencFree( &val ); tr_free( buf ); } static void parseLtepHandshake( tr_peermsgs * msgs, int len, struct evbuffer * inbuf ) { int64_t i; tr_benc val, * sub; uint8_t * tmp = tr_new( uint8_t, len ); tr_peerIoReadBytes( msgs->peer->io, inbuf, tmp, len ); msgs->peerSentLtepHandshake = 1; if( tr_bencLoad( tmp, len, &val, NULL ) || !tr_bencIsDict( &val ) ) { dbgmsg( msgs, "GET extended-handshake, couldn't get dictionary" ); tr_free( tmp ); return; } dbgmsg( msgs, "here is the handshake: [%*.*s]", len, len, tmp ); /* does the peer prefer encrypted connections? */ if( tr_bencDictFindInt( &val, "e", &i ) ) msgs->peer->encryption_preference = i ? ENCRYPTION_PREFERENCE_YES : ENCRYPTION_PREFERENCE_NO; /* check supported messages for utorrent pex */ msgs->peerSupportsPex = 0; if( tr_bencDictFindDict( &val, "m", &sub ) ) { if( tr_bencDictFindInt( sub, "ut_pex", &i ) ) { msgs->ut_pex_id = (uint8_t) i; msgs->peerSupportsPex = msgs->ut_pex_id == 0 ? 0 : 1; dbgmsg( msgs, "msgs->ut_pex is %d", (int)msgs->ut_pex_id ); } } /* look for upload_only (BEP 21) */ if( tr_bencDictFindInt( &val, "upload_only", &i ) ) fireUploadOnly( msgs, i!=0 ); /* get peer's listening port */ if( tr_bencDictFindInt( &val, "p", &i ) ) { msgs->peer->port = htons( (uint16_t)i ); dbgmsg( msgs, "msgs->port is now %hu", msgs->peer->port ); } /* get peer's maximum request queue size */ if( tr_bencDictFindInt( &val, "reqq", &i ) ) msgs->reqq = i; tr_bencFree( &val ); tr_free( tmp ); } static void parseUtPex( tr_peermsgs * msgs, int msglen, struct evbuffer * inbuf ) { int loaded = 0; uint8_t * tmp = tr_new( uint8_t, msglen ); tr_benc val; tr_torrent * tor = msgs->torrent; const uint8_t * added; size_t added_len; tr_peerIoReadBytes( msgs->peer->io, inbuf, tmp, msglen ); if( tr_torrentAllowsPex( tor ) && ( ( loaded = !tr_bencLoad( tmp, msglen, &val, NULL ) ) ) ) { if( tr_bencDictFindRaw( &val, "added", &added, &added_len ) ) { const uint8_t * added_f = NULL; tr_pex * pex; size_t i, n; size_t added_f_len = 0; tr_bencDictFindRaw( &val, "added.f", &added_f, &added_f_len ); pex = tr_peerMgrCompactToPex( added, added_len, added_f, added_f_len, &n ); for( i = 0; i < n; ++i ) tr_peerMgrAddPex( tor, TR_PEER_FROM_PEX, pex + i ); tr_free( pex ); } if( tr_bencDictFindRaw( &val, "added6", &added, &added_len ) ) { const uint8_t * added_f = NULL; tr_pex * pex; size_t i, n; size_t added_f_len = 0; tr_bencDictFindRaw( &val, "added6.f", &added_f, &added_f_len ); pex = tr_peerMgrCompact6ToPex( added, added_len, added_f, added_f_len, &n ); for( i = 0; i < n; ++i ) tr_peerMgrAddPex( tor, TR_PEER_FROM_PEX, pex + i ); tr_free( pex ); } } if( loaded ) tr_bencFree( &val ); tr_free( tmp ); } static void sendPex( tr_peermsgs * msgs ); static void parseLtep( tr_peermsgs * msgs, int msglen, struct evbuffer * inbuf ) { uint8_t ltep_msgid; tr_peerIoReadUint8( msgs->peer->io, inbuf, <ep_msgid ); msglen--; if( ltep_msgid == LTEP_HANDSHAKE ) { dbgmsg( msgs, "got ltep handshake" ); parseLtepHandshake( msgs, msglen, inbuf ); if( tr_peerIoSupportsLTEP( msgs->peer->io ) ) { sendLtepHandshake( msgs ); sendPex( msgs ); } } else if( ltep_msgid == TR_LTEP_PEX ) { dbgmsg( msgs, "got ut pex" ); msgs->peerSupportsPex = 1; parseUtPex( msgs, msglen, inbuf ); } else { dbgmsg( msgs, "skipping unknown ltep message (%d)", (int)ltep_msgid ); evbuffer_drain( inbuf, msglen ); } } static int readBtLength( tr_peermsgs * msgs, struct evbuffer * inbuf, size_t inlen ) { uint32_t len; if( inlen < sizeof( len ) ) return READ_LATER; tr_peerIoReadUint32( msgs->peer->io, inbuf, &len ); if( len == 0 ) /* peer sent us a keepalive message */ dbgmsg( msgs, "got KeepAlive" ); else { msgs->incoming.length = len; msgs->state = AWAITING_BT_ID; } return READ_NOW; } static int readBtMessage( tr_peermsgs * msgs, struct evbuffer * inbuf, size_t inlen ); static int readBtId( tr_peermsgs * msgs, struct evbuffer * inbuf, size_t inlen ) { uint8_t id; if( inlen < sizeof( uint8_t ) ) return READ_LATER; tr_peerIoReadUint8( msgs->peer->io, inbuf, &id ); msgs->incoming.id = id; dbgmsg( msgs, "msgs->incoming.id is now %d; msgs->incoming.length is %zu", id, (size_t)msgs->incoming.length ); if( id == BT_PIECE ) { msgs->state = AWAITING_BT_PIECE; return READ_NOW; } else if( msgs->incoming.length != 1 ) { msgs->state = AWAITING_BT_MESSAGE; return READ_NOW; } else return readBtMessage( msgs, inbuf, inlen - 1 ); } static void updatePeerProgress( tr_peermsgs * msgs ) { msgs->peer->progress = tr_bitfieldCountTrueBits( msgs->peer->have ) / (float)msgs->torrent->info.pieceCount; dbgmsg( msgs, "peer progress is %f", msgs->peer->progress ); updateFastSet( msgs ); updateInterest( msgs ); firePeerProgress( msgs ); } static void peerMadeRequest( tr_peermsgs * msgs, const struct peer_request * req ) { const tr_bool fext = tr_peerIoSupportsFEXT( msgs->peer->io ); const int reqIsValid = requestIsValid( msgs, req ); const int clientHasPiece = reqIsValid && tr_cpPieceIsComplete( &msgs->torrent->completion, req->index ); const int peerIsChoked = msgs->peer->peerIsChoked; int allow = FALSE; if( !reqIsValid ) dbgmsg( msgs, "rejecting an invalid request." ); else if( !clientHasPiece ) dbgmsg( msgs, "rejecting request for a piece we don't have." ); else if( peerIsChoked ) dbgmsg( msgs, "rejecting request from choked peer" ); else allow = TRUE; if( allow ) reqListAppend( &msgs->peerAskedFor, req ); else if( fext ) protocolSendReject( msgs, req ); } static tr_bool messageLengthIsCorrect( const tr_peermsgs * msg, uint8_t id, uint32_t len ) { switch( id ) { case BT_CHOKE: case BT_UNCHOKE: case BT_INTERESTED: case BT_NOT_INTERESTED: case BT_FEXT_HAVE_ALL: case BT_FEXT_HAVE_NONE: return len == 1; case BT_HAVE: case BT_FEXT_SUGGEST: case BT_FEXT_ALLOWED_FAST: return len == 5; case BT_BITFIELD: return len == ( msg->torrent->info.pieceCount + 7u ) / 8u + 1u; case BT_REQUEST: case BT_CANCEL: case BT_FEXT_REJECT: return len == 13; case BT_PIECE: return len > 9 && len <= 16393; case BT_PORT: return len == 3; case BT_LTEP: return len >= 2; default: return FALSE; } } static int clientGotBlock( tr_peermsgs * msgs, const uint8_t * block, const struct peer_request * req ); static int readBtPiece( tr_peermsgs * msgs, struct evbuffer * inbuf, size_t inlen, size_t * setme_piece_bytes_read ) { struct peer_request * req = &msgs->incoming.blockReq; assert( EVBUFFER_LENGTH( inbuf ) >= inlen ); dbgmsg( msgs, "In readBtPiece" ); if( !req->length ) { if( inlen < 8 ) return READ_LATER; tr_peerIoReadUint32( msgs->peer->io, inbuf, &req->index ); tr_peerIoReadUint32( msgs->peer->io, inbuf, &req->offset ); req->length = msgs->incoming.length - 9; dbgmsg( msgs, "got incoming block header %u:%u->%u", req->index, req->offset, req->length ); return READ_NOW; } else { int err; /* read in another chunk of data */ const size_t nLeft = req->length - EVBUFFER_LENGTH( msgs->incoming.block ); size_t n = MIN( nLeft, inlen ); size_t i = n; while( i > 0 ) { uint8_t buf[MAX_STACK_ARRAY_SIZE]; const size_t thisPass = MIN( i, sizeof( buf ) ); tr_peerIoReadBytes( msgs->peer->io, inbuf, buf, thisPass ); evbuffer_add( msgs->incoming.block, buf, thisPass ); i -= thisPass; } fireClientGotData( msgs, n, TRUE ); *setme_piece_bytes_read += n; dbgmsg( msgs, "got %zu bytes for block %u:%u->%u ... %d remain", n, req->index, req->offset, req->length, (int)( req->length - EVBUFFER_LENGTH( msgs->incoming.block ) ) ); if( EVBUFFER_LENGTH( msgs->incoming.block ) < req->length ) return READ_LATER; /* we've got the whole block ... process it */ err = clientGotBlock( msgs, EVBUFFER_DATA( msgs->incoming.block ), req ); /* cleanup */ evbuffer_free( msgs->incoming.block ); msgs->incoming.block = evbuffer_new( ); req->length = 0; msgs->state = AWAITING_BT_LENGTH; if( !err ) return READ_NOW; else { fireError( msgs, err ); return READ_ERR; } } } static int readBtMessage( tr_peermsgs * msgs, struct evbuffer * inbuf, size_t inlen ) { uint32_t ui32; uint32_t msglen = msgs->incoming.length; const uint8_t id = msgs->incoming.id; const size_t startBufLen = EVBUFFER_LENGTH( inbuf ); const tr_bool fext = tr_peerIoSupportsFEXT( msgs->peer->io ); --msglen; /* id length */ dbgmsg( msgs, "got BT id %d, len %d, buffer size is %zu", (int)id, (int)msglen, inlen ); if( inlen < msglen ) return READ_LATER; if( !messageLengthIsCorrect( msgs, id, msglen + 1 ) ) { dbgmsg( msgs, "bad packet - BT message #%d with a length of %d", (int)id, (int)msglen ); fireError( msgs, EMSGSIZE ); return READ_ERR; } switch( id ) { case BT_CHOKE: dbgmsg( msgs, "got Choke" ); msgs->peer->clientIsChoked = 1; if( !fext ) cancelAllRequestsToPeer( msgs, FALSE ); break; case BT_UNCHOKE: dbgmsg( msgs, "got Unchoke" ); msgs->peer->clientIsChoked = 0; fireNeedReq( msgs ); break; case BT_INTERESTED: dbgmsg( msgs, "got Interested" ); msgs->peer->peerIsInterested = 1; break; case BT_NOT_INTERESTED: dbgmsg( msgs, "got Not Interested" ); msgs->peer->peerIsInterested = 0; break; case BT_HAVE: tr_peerIoReadUint32( msgs->peer->io, inbuf, &ui32 ); dbgmsg( msgs, "got Have: %u", ui32 ); if( tr_bitfieldAdd( msgs->peer->have, ui32 ) ) { fireError( msgs, ERANGE ); return READ_ERR; } updatePeerProgress( msgs ); tr_rcTransferred( &msgs->torrent->swarmSpeed, msgs->torrent->info.pieceSize ); break; case BT_BITFIELD: { dbgmsg( msgs, "got a bitfield" ); tr_peerIoReadBytes( msgs->peer->io, inbuf, msgs->peer->have->bits, msglen ); updatePeerProgress( msgs ); fireNeedReq( msgs ); break; } case BT_REQUEST: { struct peer_request r; tr_peerIoReadUint32( msgs->peer->io, inbuf, &r.index ); tr_peerIoReadUint32( msgs->peer->io, inbuf, &r.offset ); tr_peerIoReadUint32( msgs->peer->io, inbuf, &r.length ); dbgmsg( msgs, "got Request: %u:%u->%u", r.index, r.offset, r.length ); peerMadeRequest( msgs, &r ); break; } case BT_CANCEL: { struct peer_request r; tr_peerIoReadUint32( msgs->peer->io, inbuf, &r.index ); tr_peerIoReadUint32( msgs->peer->io, inbuf, &r.offset ); tr_peerIoReadUint32( msgs->peer->io, inbuf, &r.length ); dbgmsg( msgs, "got a Cancel %u:%u->%u", r.index, r.offset, r.length ); if( reqListRemove( &msgs->peerAskedFor, &r ) && fext ) protocolSendReject( msgs, &r ); break; } case BT_PIECE: assert( 0 ); /* handled elsewhere! */ break; case BT_PORT: dbgmsg( msgs, "Got a BT_PORT" ); tr_peerIoReadUint16( msgs->peer->io, inbuf, &msgs->peer->dht_port ); if( msgs->peer->dht_port > 0 ) tr_dhtAddNode( getSession(msgs), &msgs->peer->addr, msgs->peer->dht_port, 0 ); break; case BT_FEXT_SUGGEST: dbgmsg( msgs, "Got a BT_FEXT_SUGGEST" ); tr_peerIoReadUint32( msgs->peer->io, inbuf, &ui32 ); if( fext ) fireClientGotSuggest( msgs, ui32 ); else { fireError( msgs, EMSGSIZE ); return READ_ERR; } break; case BT_FEXT_ALLOWED_FAST: dbgmsg( msgs, "Got a BT_FEXT_ALLOWED_FAST" ); tr_peerIoReadUint32( msgs->peer->io, inbuf, &ui32 ); if( fext ) fireClientGotAllowedFast( msgs, ui32 ); else { fireError( msgs, EMSGSIZE ); return READ_ERR; } break; case BT_FEXT_HAVE_ALL: dbgmsg( msgs, "Got a BT_FEXT_HAVE_ALL" ); if( fext ) { tr_bitfieldAddRange( msgs->peer->have, 0, msgs->torrent->info.pieceCount ); updatePeerProgress( msgs ); } else { fireError( msgs, EMSGSIZE ); return READ_ERR; } break; case BT_FEXT_HAVE_NONE: dbgmsg( msgs, "Got a BT_FEXT_HAVE_NONE" ); if( fext ) { tr_bitfieldClear( msgs->peer->have ); updatePeerProgress( msgs ); } else { fireError( msgs, EMSGSIZE ); return READ_ERR; } break; case BT_FEXT_REJECT: { struct peer_request r; dbgmsg( msgs, "Got a BT_FEXT_REJECT" ); tr_peerIoReadUint32( msgs->peer->io, inbuf, &r.index ); tr_peerIoReadUint32( msgs->peer->io, inbuf, &r.offset ); tr_peerIoReadUint32( msgs->peer->io, inbuf, &r.length ); if( fext ) reqListRemove( &msgs->clientAskedFor, &r ); else { fireError( msgs, EMSGSIZE ); return READ_ERR; } break; } case BT_LTEP: dbgmsg( msgs, "Got a BT_LTEP" ); parseLtep( msgs, msglen, inbuf ); break; default: dbgmsg( msgs, "peer sent us an UNKNOWN: %d", (int)id ); tr_peerIoDrain( msgs->peer->io, inbuf, msglen ); break; } assert( msglen + 1 == msgs->incoming.length ); assert( EVBUFFER_LENGTH( inbuf ) == startBufLen - msglen ); msgs->state = AWAITING_BT_LENGTH; return READ_NOW; } static TR_INLINE void decrementDownloadedCount( tr_peermsgs * msgs, uint32_t byteCount ) { tr_torrent * tor = msgs->torrent; tor->downloadedCur -= MIN( tor->downloadedCur, byteCount ); } static TR_INLINE void clientGotUnwantedBlock( tr_peermsgs * msgs, const struct peer_request * req ) { decrementDownloadedCount( msgs, req->length ); } static void addPeerToBlamefield( tr_peermsgs * msgs, uint32_t index ) { if( !msgs->peer->blame ) msgs->peer->blame = tr_bitfieldNew( msgs->torrent->info.pieceCount ); tr_bitfieldAdd( msgs->peer->blame, index ); } /* returns 0 on success, or an errno on failure */ static int clientGotBlock( tr_peermsgs * msgs, const uint8_t * data, const struct peer_request * req ) { int err; tr_torrent * tor = msgs->torrent; const tr_block_index_t block = _tr_block( tor, req->index, req->offset ); assert( msgs ); assert( req ); if( req->length != tr_torBlockCountBytes( msgs->torrent, block ) ) { dbgmsg( msgs, "wrong block size -- expected %u, got %d", tr_torBlockCountBytes( msgs->torrent, block ), req->length ); return EMSGSIZE; } /* save the block */ dbgmsg( msgs, "got block %u:%u->%u", req->index, req->offset, req->length ); /** *** Remove the block from our `we asked for this' list **/ if( !reqListRemove( &msgs->clientAskedFor, req ) ) { clientGotUnwantedBlock( msgs, req ); dbgmsg( msgs, "we didn't ask for this message..." ); return 0; } dbgmsg( msgs, "peer has %d more blocks we've asked for", msgs->clientAskedFor.len ); /** *** Error checks **/ if( tr_cpBlockIsComplete( &tor->completion, block ) ) { dbgmsg( msgs, "we have this block already..." ); clientGotUnwantedBlock( msgs, req ); return 0; } /** *** Save the block **/ if(( err = tr_ioWrite( tor, req->index, req->offset, req->length, data ))) return err; addPeerToBlamefield( msgs, req->index ); fireGotBlock( msgs, req ); return 0; } static int peerPulse( void * vmsgs ); static void didWrite( tr_peerIo * io UNUSED, size_t bytesWritten, int wasPieceData, void * vmsgs ) { tr_peermsgs * msgs = vmsgs; firePeerGotData( msgs, bytesWritten, wasPieceData ); if ( tr_isPeerIo( io ) && io->userData ) peerPulse( msgs ); } static ReadState canRead( tr_peerIo * io, void * vmsgs, size_t * piece ) { ReadState ret; tr_peermsgs * msgs = vmsgs; struct evbuffer * in = tr_peerIoGetReadBuffer( io ); const size_t inlen = EVBUFFER_LENGTH( in ); dbgmsg( msgs, "canRead: inlen is %zu, msgs->state is %d", inlen, msgs->state ); if( !inlen ) { ret = READ_LATER; } else if( msgs->state == AWAITING_BT_PIECE ) { ret = inlen ? readBtPiece( msgs, in, inlen, piece ) : READ_LATER; } else switch( msgs->state ) { case AWAITING_BT_LENGTH: ret = readBtLength ( msgs, in, inlen ); break; case AWAITING_BT_ID: ret = readBtId ( msgs, in, inlen ); break; case AWAITING_BT_MESSAGE: ret = readBtMessage( msgs, in, inlen ); break; default: ret = READ_ERR; assert( 0 ); } dbgmsg( msgs, "canRead: ret is %d", (int)ret ); /* log the raw data that was read */ if( ( ret != READ_ERR ) && ( EVBUFFER_LENGTH( in ) != inlen ) ) fireClientGotData( msgs, inlen - EVBUFFER_LENGTH( in ), FALSE ); return ret; } /** *** **/ static int ratePulse( tr_peermsgs * msgs, uint64_t now ) { const double rateToClient = tr_peerGetPieceSpeed( msgs->peer, now, TR_PEER_TO_CLIENT ); const int seconds = 10; const int floor = 8; const int estimatedBlocksInPeriod = ( rateToClient * seconds * 1024 ) / msgs->torrent->blockSize; msgs->maxActiveRequests = floor + estimatedBlocksInPeriod; if( msgs->reqq > 0 ) msgs->maxActiveRequests = MIN( msgs->maxActiveRequests, msgs->reqq ); return TRUE; } static size_t fillOutputBuffer( tr_peermsgs * msgs, time_t now ) { size_t bytesWritten = 0; struct peer_request req; const tr_bool haveMessages = EVBUFFER_LENGTH( msgs->outMessages ) != 0; const tr_bool fext = tr_peerIoSupportsFEXT( msgs->peer->io ); /** *** Protocol messages **/ if( haveMessages && !msgs->outMessagesBatchedAt ) /* fresh batch */ { dbgmsg( msgs, "started an outMessages batch (length is %zu)", EVBUFFER_LENGTH( msgs->outMessages ) ); msgs->outMessagesBatchedAt = now; } else if( haveMessages && ( ( now - msgs->outMessagesBatchedAt ) >= msgs->outMessagesBatchPeriod ) ) { const size_t len = EVBUFFER_LENGTH( msgs->outMessages ); /* flush the protocol messages */ dbgmsg( msgs, "flushing outMessages... to %p (length is %zu)", msgs->peer->io, len ); tr_peerIoWriteBuf( msgs->peer->io, msgs->outMessages, FALSE ); msgs->clientSentAnythingAt = now; msgs->outMessagesBatchedAt = 0; msgs->outMessagesBatchPeriod = LOW_PRIORITY_INTERVAL_SECS; bytesWritten += len; } /** *** Blocks **/ if( ( tr_peerIoGetWriteBufferSpace( msgs->peer->io, now ) >= msgs->torrent->blockSize ) && popNextRequest( msgs, &req ) ) { if( requestIsValid( msgs, &req ) && tr_cpPieceIsComplete( &msgs->torrent->completion, req.index ) ) { /* FIXME(libevent2) use evbuffer_reserve_space() + evbuffer_commit_space() */ int err; const uint32_t msglen = 4 + 1 + 4 + 4 + req.length; struct evbuffer * out; tr_peerIo * io = msgs->peer->io; out = evbuffer_new( ); evbuffer_expand( out, msglen ); tr_peerIoWriteUint32( io, out, sizeof( uint8_t ) + 2 * sizeof( uint32_t ) + req.length ); tr_peerIoWriteUint8 ( io, out, BT_PIECE ); tr_peerIoWriteUint32( io, out, req.index ); tr_peerIoWriteUint32( io, out, req.offset ); err = tr_ioRead( msgs->torrent, req.index, req.offset, req.length, EVBUFFER_DATA(out)+EVBUFFER_LENGTH(out) ); if( err ) { if( fext ) protocolSendReject( msgs, &req ); } else { dbgmsg( msgs, "sending block %u:%u->%u", req.index, req.offset, req.length ); EVBUFFER_LENGTH(out) += req.length; assert( EVBUFFER_LENGTH( out ) == msglen ); tr_peerIoWriteBuf( io, out, TRUE ); bytesWritten += EVBUFFER_LENGTH( out ); msgs->clientSentAnythingAt = now; } evbuffer_free( out ); if( err ) { fireError( msgs, err ); bytesWritten = 0; msgs = NULL; } } else if( fext ) /* peer needs a reject message */ { protocolSendReject( msgs, &req ); } } /** *** Keepalive **/ if( ( msgs != NULL ) && ( msgs->clientSentAnythingAt != 0 ) && ( ( now - msgs->clientSentAnythingAt ) > KEEPALIVE_INTERVAL_SECS ) ) { dbgmsg( msgs, "sending a keepalive message" ); tr_peerIoWriteUint32( msgs->peer->io, msgs->outMessages, 0 ); pokeBatchPeriod( msgs, IMMEDIATE_PRIORITY_INTERVAL_SECS ); } return bytesWritten; } static int peerPulse( void * vmsgs ) { tr_peermsgs * msgs = vmsgs; const time_t now = time( NULL ); ratePulse( msgs, now ); pumpRequestQueue( msgs, now ); expireOldRequests( msgs, now ); for( ;; ) if( fillOutputBuffer( msgs, now ) < 1 ) break; return TRUE; /* loop forever */ } void tr_peerMsgsPulse( tr_peermsgs * msgs ) { if( msgs != NULL ) peerPulse( msgs ); } static void gotError( tr_peerIo * io UNUSED, short what, void * vmsgs ) { if( what & EVBUFFER_TIMEOUT ) dbgmsg( vmsgs, "libevent got a timeout, what=%hd", what ); if( what & ( EVBUFFER_EOF | EVBUFFER_ERROR ) ) dbgmsg( vmsgs, "libevent got an error! what=%hd, errno=%d (%s)", what, errno, tr_strerror( errno ) ); fireError( vmsgs, ENOTCONN ); } static void sendBitfield( tr_peermsgs * msgs ) { struct evbuffer * out = msgs->outMessages; tr_bitfield * field; tr_piece_index_t lazyPieces[LAZY_PIECE_COUNT]; size_t i; size_t lazyCount = 0; field = tr_bitfieldDup( tr_cpPieceBitfield( &msgs->torrent->completion ) ); if( tr_sessionIsLazyBitfieldEnabled( getSession( msgs ) ) ) { /** Lazy bitfields aren't a high priority or secure, so I'm opting for speed over a truly random sample -- let's limit the pool size to the first 1000 pieces so large torrents don't bog things down */ size_t poolSize; const size_t maxPoolSize = MIN( msgs->torrent->info.pieceCount, 1000 ); tr_piece_index_t * pool = tr_new( tr_piece_index_t, maxPoolSize ); /* build the pool */ for( i=poolSize=0; i 0 ) && ( lazyCount < LAZY_PIECE_COUNT ) ) { const int pos = tr_cryptoWeakRandInt( poolSize ); const tr_piece_index_t piece = pool[pos]; tr_bitfieldRem( field, piece ); lazyPieces[lazyCount++] = piece; pool[pos] = pool[--poolSize]; } /* cleanup */ tr_free( pool ); } tr_peerIoWriteUint32( msgs->peer->io, out, sizeof( uint8_t ) + field->byteCount ); tr_peerIoWriteUint8 ( msgs->peer->io, out, BT_BITFIELD ); /* FIXME(libevent2): use evbuffer_add_reference() */ tr_peerIoWriteBytes ( msgs->peer->io, out, field->bits, field->byteCount ); dbgmsg( msgs, "sending bitfield... outMessage size is now %zu", EVBUFFER_LENGTH( out ) ); pokeBatchPeriod( msgs, IMMEDIATE_PRIORITY_INTERVAL_SECS ); for( i = 0; i < lazyCount; ++i ) protocolSendHave( msgs, lazyPieces[i] ); tr_bitfieldFree( field ); } static void tellPeerWhatWeHave( tr_peermsgs * msgs ) { const tr_bool fext = tr_peerIoSupportsFEXT( msgs->peer->io ); if( fext && ( tr_cpGetStatus( &msgs->torrent->completion ) == TR_SEED ) ) { protocolSendHaveAll( msgs ); } else if( fext && ( tr_cpHaveValid( &msgs->torrent->completion ) == 0 ) ) { protocolSendHaveNone( msgs ); } else { sendBitfield( msgs ); } } /** *** **/ /* some peers give us error messages if we send more than this many peers in a single pex message http://wiki.theory.org/BitTorrentPeerExchangeConventions */ #define MAX_PEX_ADDED 50 #define MAX_PEX_DROPPED 50 typedef struct { tr_pex * added; tr_pex * dropped; tr_pex * elements; int addedCount; int droppedCount; int elementCount; } PexDiffs; static void pexAddedCb( void * vpex, void * userData ) { PexDiffs * diffs = userData; tr_pex * pex = vpex; if( diffs->addedCount < MAX_PEX_ADDED ) { diffs->added[diffs->addedCount++] = *pex; diffs->elements[diffs->elementCount++] = *pex; } } static TR_INLINE void pexDroppedCb( void * vpex, void * userData ) { PexDiffs * diffs = userData; tr_pex * pex = vpex; if( diffs->droppedCount < MAX_PEX_DROPPED ) { diffs->dropped[diffs->droppedCount++] = *pex; } } static TR_INLINE void pexElementCb( void * vpex, void * userData ) { PexDiffs * diffs = userData; tr_pex * pex = vpex; diffs->elements[diffs->elementCount++] = *pex; } static void sendPex( tr_peermsgs * msgs ) { if( msgs->peerSupportsPex && tr_torrentAllowsPex( msgs->torrent ) ) { PexDiffs diffs; PexDiffs diffs6; tr_pex * newPex = NULL; tr_pex * newPex6 = NULL; const int newCount = tr_peerMgrGetPeers( msgs->torrent, &newPex, TR_AF_INET ); const int newCount6 = tr_peerMgrGetPeers( msgs->torrent, &newPex6, TR_AF_INET6 ); /* build the diffs */ diffs.added = tr_new( tr_pex, newCount ); diffs.addedCount = 0; diffs.dropped = tr_new( tr_pex, msgs->pexCount ); diffs.droppedCount = 0; diffs.elements = tr_new( tr_pex, newCount + msgs->pexCount ); diffs.elementCount = 0; tr_set_compare( msgs->pex, msgs->pexCount, newPex, newCount, tr_pexCompare, sizeof( tr_pex ), pexDroppedCb, pexAddedCb, pexElementCb, &diffs ); diffs6.added = tr_new( tr_pex, newCount6 ); diffs6.addedCount = 0; diffs6.dropped = tr_new( tr_pex, msgs->pexCount6 ); diffs6.droppedCount = 0; diffs6.elements = tr_new( tr_pex, newCount6 + msgs->pexCount6 ); diffs6.elementCount = 0; tr_set_compare( msgs->pex6, msgs->pexCount6, newPex6, newCount6, tr_pexCompare, sizeof( tr_pex ), pexDroppedCb, pexAddedCb, pexElementCb, &diffs6 ); dbgmsg( msgs, "pex: old peer count %d, new peer count %d, added %d, removed %d", msgs->pexCount, newCount + newCount6, diffs.addedCount + diffs6.addedCount, diffs.droppedCount + diffs6.droppedCount ); if( !diffs.addedCount && !diffs.droppedCount && !diffs6.addedCount && !diffs6.droppedCount ) { tr_free( diffs.elements ); tr_free( diffs6.elements ); } else { int i; tr_benc val; char * benc; int bencLen; uint8_t * tmp, *walk; tr_peerIo * io = msgs->peer->io; struct evbuffer * out = msgs->outMessages; /* update peer */ tr_free( msgs->pex ); msgs->pex = diffs.elements; msgs->pexCount = diffs.elementCount; tr_free( msgs->pex6 ); msgs->pex6 = diffs6.elements; msgs->pexCount6 = diffs6.elementCount; /* build the pex payload */ tr_bencInitDict( &val, 3 ); /* ipv6 support: left as 3: * speed vs. likelihood? */ /* "added" */ tmp = walk = tr_new( uint8_t, diffs.addedCount * 6 ); for( i = 0; i < diffs.addedCount; ++i ) { memcpy( walk, &diffs.added[i].addr.addr, 4 ); walk += 4; memcpy( walk, &diffs.added[i].port, 2 ); walk += 2; } assert( ( walk - tmp ) == diffs.addedCount * 6 ); tr_bencDictAddRaw( &val, "added", tmp, walk - tmp ); tr_free( tmp ); /* "added.f" */ tmp = walk = tr_new( uint8_t, diffs.addedCount ); for( i = 0; i < diffs.addedCount; ++i ) *walk++ = diffs.added[i].flags; assert( ( walk - tmp ) == diffs.addedCount ); tr_bencDictAddRaw( &val, "added.f", tmp, walk - tmp ); tr_free( tmp ); /* "dropped" */ tmp = walk = tr_new( uint8_t, diffs.droppedCount * 6 ); for( i = 0; i < diffs.droppedCount; ++i ) { memcpy( walk, &diffs.dropped[i].addr.addr, 4 ); walk += 4; memcpy( walk, &diffs.dropped[i].port, 2 ); walk += 2; } assert( ( walk - tmp ) == diffs.droppedCount * 6 ); tr_bencDictAddRaw( &val, "dropped", tmp, walk - tmp ); tr_free( tmp ); /* "added6" */ tmp = walk = tr_new( uint8_t, diffs6.addedCount * 18 ); for( i = 0; i < diffs6.addedCount; ++i ) { memcpy( walk, &diffs6.added[i].addr.addr.addr6.s6_addr, 16 ); walk += 16; memcpy( walk, &diffs6.added[i].port, 2 ); walk += 2; } assert( ( walk - tmp ) == diffs6.addedCount * 18 ); tr_bencDictAddRaw( &val, "added6", tmp, walk - tmp ); tr_free( tmp ); /* "added6.f" */ tmp = walk = tr_new( uint8_t, diffs6.addedCount ); for( i = 0; i < diffs6.addedCount; ++i ) *walk++ = diffs6.added[i].flags; assert( ( walk - tmp ) == diffs6.addedCount ); tr_bencDictAddRaw( &val, "added6.f", tmp, walk - tmp ); tr_free( tmp ); /* "dropped6" */ tmp = walk = tr_new( uint8_t, diffs6.droppedCount * 18 ); for( i = 0; i < diffs6.droppedCount; ++i ) { memcpy( walk, &diffs6.dropped[i].addr.addr.addr6.s6_addr, 16 ); walk += 16; memcpy( walk, &diffs6.dropped[i].port, 2 ); walk += 2; } assert( ( walk - tmp ) == diffs6.droppedCount * 18); tr_bencDictAddRaw( &val, "dropped6", tmp, walk - tmp ); tr_free( tmp ); /* write the pex message */ benc = tr_bencToStr( &val, TR_FMT_BENC, &bencLen ); tr_peerIoWriteUint32( io, out, 2 * sizeof( uint8_t ) + bencLen ); tr_peerIoWriteUint8 ( io, out, BT_LTEP ); tr_peerIoWriteUint8 ( io, out, msgs->ut_pex_id ); tr_peerIoWriteBytes ( io, out, benc, bencLen ); pokeBatchPeriod( msgs, HIGH_PRIORITY_INTERVAL_SECS ); dbgmsg( msgs, "sending a pex message; outMessage size is now %zu", EVBUFFER_LENGTH( out ) ); dbgOutMessageLen( msgs ); tr_free( benc ); tr_bencFree( &val ); } /* cleanup */ tr_free( diffs.added ); tr_free( diffs.dropped ); tr_free( newPex ); tr_free( diffs6.added ); tr_free( diffs6.dropped ); tr_free( newPex6 ); /*msgs->clientSentPexAt = time( NULL );*/ } } static void pexPulse( int foo UNUSED, short bar UNUSED, void * vmsgs ) { struct tr_peermsgs * msgs = vmsgs; sendPex( msgs ); tr_timerAdd( &msgs->pexTimer, PEX_INTERVAL_SECS, 0 ); } /** *** **/ tr_peermsgs* tr_peerMsgsNew( struct tr_torrent * torrent, struct tr_peer * peer, tr_delivery_func func, void * userData, tr_publisher_tag * setme ) { tr_peermsgs * m; assert( peer ); assert( peer->io ); m = tr_new0( tr_peermsgs, 1 ); m->publisher = TR_PUBLISHER_INIT; m->peer = peer; m->torrent = torrent; m->peer->clientIsChoked = 1; m->peer->peerIsChoked = 1; m->peer->clientIsInterested = 0; m->peer->peerIsInterested = 0; m->peer->have = tr_bitfieldNew( torrent->info.pieceCount ); m->state = AWAITING_BT_LENGTH; m->outMessages = evbuffer_new( ); m->outMessagesBatchedAt = 0; m->outMessagesBatchPeriod = LOW_PRIORITY_INTERVAL_SECS; m->incoming.block = evbuffer_new( ); m->peerAskedFor = REQUEST_LIST_INIT; m->clientAskedFor = REQUEST_LIST_INIT; m->clientWillAskFor = REQUEST_LIST_INIT; evtimer_set( &m->pexTimer, pexPulse, m ); tr_timerAdd( &m->pexTimer, PEX_INTERVAL_SECS, 0 ); peer->msgs = m; *setme = tr_publisherSubscribe( &m->publisher, func, userData ); if( tr_peerIoSupportsLTEP( peer->io ) ) sendLtepHandshake( m ); if(tr_peerIoSupportsDHT(peer->io)) protocolSendPort(m, tr_dhtPort(torrent->session)); tellPeerWhatWeHave( m ); tr_peerIoSetIOFuncs( m->peer->io, canRead, didWrite, gotError, m ); ratePulse( m, tr_date() ); return m; } void tr_peerMsgsFree( tr_peermsgs* msgs ) { if( msgs ) { evtimer_del( &msgs->pexTimer ); tr_publisherDestruct( &msgs->publisher ); reqListClear( &msgs->clientWillAskFor ); reqListClear( &msgs->clientAskedFor ); reqListClear( &msgs->peerAskedFor ); evbuffer_free( msgs->incoming.block ); evbuffer_free( msgs->outMessages ); tr_free( msgs->pex6 ); tr_free( msgs->pex ); memset( msgs, ~0, sizeof( tr_peermsgs ) ); tr_free( msgs ); } } void tr_peerMsgsUnsubscribe( tr_peermsgs * peer, tr_publisher_tag tag ) { tr_publisherUnsubscribe( &peer->publisher, tag ); }