Index: libtransmission/peer-io.c =================================================================== --- libtransmission/peer-io.c (revision 6743) +++ libtransmission/peer-io.c (working copy) @@ -11,6 +11,7 @@ */ #include +#include /* INT_MAX */ #include #include #include @@ -34,13 +35,20 @@ #define IO_TIMEOUT_SECS 8 -/* the size of a typical request message */ -#define TR_RDBUF ((1024*16) + 13) - /** *** **/ +#define dbgmsg(io, fmt...) \ + tr_deepLog( __FILE__,__LINE__, tr_peerIoGetAddrStr(io), ##fmt ) + +struct tr_bandwidth +{ + unsigned int isUnlimited : 1; + size_t bytesUsed; + size_t bytesLeft; +}; + struct tr_peerIo { unsigned int isEncrypted : 1; @@ -57,16 +65,23 @@ struct tr_peerIo uint8_t peerId[20]; time_t timeCreated; - struct tr_handle * handle; + tr_session * session; struct in_addr in_addr; struct bufferevent * bufev; + struct evbuffer * output; tr_can_read_cb canRead; tr_did_write_cb didWrite; tr_net_error_cb gotError; void * userData; + size_t inputBufferSize; + size_t outputBufferSize; + + struct tr_bandwidth bandwidth[2]; + tr_ratecontrol * speedometer[2]; + tr_crypto * crypto; }; @@ -75,41 +90,148 @@ struct tr_peerIo **/ static void -didWriteWrapper( struct bufferevent * e, void * userData ) +adjustOutputBuffer( tr_peerIo * io ) +{ + struct evbuffer * live = EVBUFFER_OUTPUT( io->bufev ); + + if( io->bandwidth[TR_UP].isUnlimited ) + { + bufferevent_write_buffer( io->bufev, io->output ); + } + else if( io->bandwidth[TR_UP].bytesLeft > EVBUFFER_LENGTH( live ) ) + { + /* there's free space in bufev's output buffer; + try to fill it up */ + const size_t desiredLength = io->bandwidth[TR_UP].bytesLeft; + const size_t under = desiredLength - EVBUFFER_LENGTH( live ); + const size_t n = MIN( under, EVBUFFER_LENGTH( io->output ) ); + bufferevent_write( io->bufev, EVBUFFER_DATA( io->output ), n ); + evbuffer_drain( io->output, n ); + } + else if( io->bandwidth[TR_UP].bytesLeft < EVBUFFER_LENGTH( live ) ) + { + /* bufev's output buffer exceeds our bandwidth allocation; + move the excess out of bufev so it can't be sent yet */ + const size_t desiredLength = io->bandwidth[TR_UP].bytesLeft; + const size_t over = EVBUFFER_LENGTH( live ) - desiredLength; + struct evbuffer * buf = evbuffer_new( ); + evbuffer_add( buf, EVBUFFER_DATA( live ) + desiredLength, over ); + evbuffer_add_buffer( buf, io->output ); + evbuffer_free( io->output ); + io->output = buf; + EVBUFFER_LENGTH( live ) = desiredLength; + } + else if( EVBUFFER_LENGTH( live ) ) + { + bufferevent_enable( io->bufev, EV_WRITE ); + } + + io->outputBufferSize = EVBUFFER_LENGTH( live ); + + dbgmsg( io, "after adjusting the output buffer, its size is now %zu", + io->outputBufferSize ); +} + +static void +adjustInputBuffer( tr_peerIo * io ) +{ + if( io->bandwidth[TR_DOWN].isUnlimited ) + { + dbgmsg( io, "unlimited reading..." ); + bufferevent_setwatermark( io->bufev, EV_READ, 0, 0 ); + bufferevent_enable( io->bufev, EV_READ ); + } + else + { + const size_t n = io->bandwidth[TR_DOWN].bytesLeft; + if( n == 0 ) + { + dbgmsg( io, "disabling reads because we've hit our limit" ); + bufferevent_disable( io->bufev, EV_READ ); + } + else + { + dbgmsg( io, "enabling reading of %zu more bytes", n ); + bufferevent_setwatermark( io->bufev, EV_READ, 0, n ); + bufferevent_enable( io->bufev, EV_READ ); + } + } +} + +/*** +**** +***/ + +static void +didWriteWrapper( struct bufferevent * e, int n, void * vio ) { - tr_peerIo * c = (tr_peerIo *) userData; - if( c->didWrite ) - c->didWrite( e, c->userData ); + tr_peerIo * io = vio; + + dbgmsg( io, "didWrite... io->outputBufferSize was %zu, is now %zu", + io->outputBufferSize, EVBUFFER_LENGTH( e->output ) ); + + if( n > 0 ) + { + struct tr_bandwidth * b = &io->bandwidth[TR_UP]; + b->bytesLeft -= MIN( b->bytesLeft, (size_t)n ); + b->bytesUsed += n; + tr_rcTransferred( io->speedometer[TR_UP], n ); + dbgmsg( io, "wrote %d bytes to peer... upload bytesLeft is now %zu", + n, b->bytesLeft ); + } + + adjustOutputBuffer( io ); + + if( io->didWrite ) + io->didWrite( e, io->userData ); } static void -canReadWrapper( struct bufferevent * e, void * userData ) +canReadWrapper( struct bufferevent * e, int n, void * vio ) { int done = 0; - tr_peerIo * c = userData; - tr_handle * handle = c->handle; + tr_peerIo * io = vio; + tr_session * session = io->session; + + dbgmsg( io, "canRead" ); - if( c->canRead == NULL ) - return; + /* if the input buffer has grown, record the bytes that were read */ + if( n > 0 ) + { + struct tr_bandwidth * b = io->bandwidth + TR_DOWN; + b->bytesLeft -= MIN( b->bytesLeft, (size_t)n ); + b->bytesUsed += n; + tr_rcTransferred( io->speedometer[TR_DOWN], n ); + dbgmsg( io, "%d new input bytes. bytesUsed is %zu, bytesLeft is %zu", + n, b->bytesUsed, b->bytesLeft ); - tr_globalLock( handle ); + adjustInputBuffer( io ); + } - while( !done ) + /* try to consume the input buffer */ + if( io->canRead ) { - const int ret = c->canRead( e, c->userData ); + tr_globalLock( session ); - switch( ret ) + while( !done ) { - case READ_AGAIN: - if( EVBUFFER_LENGTH( e->input ) ) - continue; - case READ_MORE: - case READ_DONE: - done = 1; + const int ret = io->canRead( e, io->userData ); + + switch( ret ) + { + case READ_AGAIN: + if( EVBUFFER_LENGTH( e->input ) ) + continue; + case READ_MORE: + case READ_DONE: + done = 1; + } } + + tr_globalUnlock( session ); } - tr_globalUnlock( handle ); + io->inputBufferSize = EVBUFFER_LENGTH( e->input ); } static void @@ -124,66 +246,79 @@ gotErrorWrapper( struct bufferevent * e, *** **/ -void bufferevent_setwatermark(struct bufferevent *, short, size_t, size_t); +static void +bufevNew( tr_peerIo * io ) +{ + io->bufev = bufferevent_new( io->socket, + canReadWrapper, + didWriteWrapper, + gotErrorWrapper, + io ); + + /* tell libevent to call didWriteWrapper after every write, + * not just when the write buffer is empty */ + bufferevent_setwatermark( io->bufev, EV_WRITE, INT_MAX, 0 ); + + bufferevent_settimeout( io->bufev, io->timeout, io->timeout ); + + bufferevent_enable( io->bufev, EV_READ|EV_WRITE ); +} static tr_peerIo* -tr_peerIoNew( struct tr_handle * handle, - const struct in_addr * in_addr, - uint16_t port, - const uint8_t * torrentHash, - int isIncoming, - int socket ) +tr_peerIoNew( tr_session * session, + const struct in_addr * in_addr, + uint16_t port, + const uint8_t * torrentHash, + int isIncoming, + int socket ) { - tr_peerIo * c; + tr_peerIo * io; if( socket >= 0 ) - tr_netSetTOS( socket, handle->peerSocketTOS ); + tr_netSetTOS( socket, session->peerSocketTOS ); - c = tr_new0( tr_peerIo, 1 ); - c->crypto = tr_cryptoNew( torrentHash, isIncoming ); - c->handle = handle; - c->in_addr = *in_addr; - c->port = port; - c->socket = socket; - c->isIncoming = isIncoming ? 1 : 0; - c->timeout = IO_TIMEOUT_SECS; - c->timeCreated = time( NULL ); - c->bufev = bufferevent_new( c->socket, - canReadWrapper, - didWriteWrapper, - gotErrorWrapper, - c ); - bufferevent_settimeout( c->bufev, c->timeout, c->timeout ); - bufferevent_enable( c->bufev, EV_READ|EV_WRITE ); - bufferevent_setwatermark( c->bufev, EV_READ, 0, TR_RDBUF ); - - return c; + io = tr_new0( tr_peerIo, 1 ); + io->crypto = tr_cryptoNew( torrentHash, isIncoming ); + io->session = session; + io->in_addr = *in_addr; + io->port = port; + io->socket = socket; + io->isIncoming = isIncoming != 0; + io->timeout = IO_TIMEOUT_SECS; + io->timeCreated = time( NULL ); + io->output = evbuffer_new( ); + io->bandwidth[TR_UP].isUnlimited = 1; + io->bandwidth[TR_DOWN].isUnlimited = 1; + io->speedometer[TR_UP] = tr_rcInit( ); + io->speedometer[TR_DOWN] = tr_rcInit( ); + bufevNew( io ); + return io; } tr_peerIo* -tr_peerIoNewIncoming( struct tr_handle * handle, +tr_peerIoNewIncoming( tr_session * session, const struct in_addr * in_addr, uint16_t port, int socket ) { - assert( handle ); + assert( session ); assert( in_addr ); assert( socket >= 0 ); - return tr_peerIoNew( handle, in_addr, port, + return tr_peerIoNew( session, in_addr, port, NULL, 1, socket ); } tr_peerIo* -tr_peerIoNewOutgoing( struct tr_handle * handle, +tr_peerIoNewOutgoing( tr_session * session, const struct in_addr * in_addr, int port, const uint8_t * torrentHash ) { int socket; - assert( handle ); + assert( session ); assert( in_addr ); assert( port >= 0 ); assert( torrentHash ); @@ -192,7 +327,7 @@ tr_peerIoNewOutgoing( struct tr_handle return socket < 0 ? NULL - : tr_peerIoNew( handle, in_addr, port, torrentHash, 0, socket ); + : tr_peerIoNew( session, in_addr, port, torrentHash, 0, socket ); } static void @@ -200,6 +335,9 @@ io_dtor( void * vio ) { tr_peerIo * io = vio; + tr_rcClose( io->speedometer[TR_DOWN] ); + tr_rcClose( io->speedometer[TR_UP] ); + evbuffer_free( io->output ); bufferevent_free( io->bufev ); tr_netClose( io->socket ); tr_cryptoFree( io->crypto ); @@ -214,17 +352,17 @@ tr_peerIoFree( tr_peerIo * io ) io->canRead = NULL; io->didWrite = NULL; io->gotError = NULL; - tr_runInEventThread( io->handle, io_dtor, io ); + tr_runInEventThread( io->session, io_dtor, io ); } } -tr_handle* -tr_peerIoGetHandle( tr_peerIo * io ) +tr_session* +tr_peerIoGetSession( tr_peerIo * io ) { assert( io ); - assert( io->handle ); + assert( io->session ); - return io->handle; + return io->session; } const struct in_addr* @@ -256,7 +394,7 @@ void tr_peerIoTryRead( tr_peerIo * io ) { if( EVBUFFER_LENGTH( io->bufev->input ) ) - canReadWrapper( io->bufev, io ); + canReadWrapper( io->bufev, 0, io ); } void @@ -292,19 +430,10 @@ tr_peerIoReconnect( tr_peerIo * io ) if( io->socket >= 0 ) { - tr_netSetTOS( io->socket, io->handle->peerSocketTOS ); + tr_netSetTOS( io->socket, io->session->peerSocketTOS ); bufferevent_free( io->bufev ); - - io->bufev = bufferevent_new( io->socket, - canReadWrapper, - didWriteWrapper, - gotErrorWrapper, - io ); - bufferevent_settimeout( io->bufev, io->timeout, io->timeout ); - bufferevent_enable( io->bufev, EV_READ|EV_WRITE ); - bufferevent_setwatermark( io->bufev, EV_READ, 0, TR_RDBUF ); - + bufevNew( io ); return 0; } @@ -412,34 +541,103 @@ tr_peerIoSupportsFEXT( const tr_peerIo * return io->fastPeersSupported; } + /** *** **/ size_t -tr_peerIoWriteBytesWaiting( const tr_peerIo * io ) +tr_peerIoGetBandwidthUsed( const tr_peerIo * io, + tr_direction direction ) { - return EVBUFFER_LENGTH( EVBUFFER_OUTPUT( io->bufev ) ); + assert( io ); + assert( direction==TR_UP || direction==TR_DOWN ); + return io->bandwidth[direction].bytesUsed; } - + +size_t +tr_peerIoGetBandwidthLeft( const tr_peerIo * io, + tr_direction direction ) +{ + assert( io ); + assert( direction==TR_UP || direction==TR_DOWN ); + return io->bandwidth[direction].bytesLeft; +} + +size_t +tr_peerIoGetWriteBufferSpace( const tr_peerIo * io ) +{ + const size_t desiredBufferLen = 4096; + const size_t currentLiveLen = EVBUFFER_LENGTH( EVBUFFER_OUTPUT( io->bufev ) ); + + const size_t desiredLiveLen = tr_peerIoGetBandwidthLeft( io, TR_UP ); + const size_t currentLbufLen = EVBUFFER_LENGTH( io->output ); + + const size_t desiredLen = desiredBufferLen + desiredLiveLen; + const size_t currentLen = currentLiveLen + currentLbufLen; + + size_t freeSpace = 0; + + if( desiredLen > currentLen ) + freeSpace = desiredLen - currentLen; + else + freeSpace = 0; + +fprintf( stderr, "peer %p write buffer space: %zu\n", io, freeSpace ); + + return freeSpace; +} + void -tr_peerIoWrite( tr_peerIo * io, - const void * writeme, - int writeme_len ) +tr_peerIoSetBandwidth( tr_peerIo * io, + tr_direction direction, + size_t bytesLeft ) { - assert( tr_amInEventThread( io->handle ) ); - bufferevent_write( io->bufev, writeme, writeme_len ); + struct tr_bandwidth * b; + + assert( io ); + assert( direction==TR_UP || direction==TR_DOWN ); + + b = io->bandwidth + direction; + b->isUnlimited = 0; + b->bytesUsed = 0; + b->bytesLeft = bytesLeft; + + adjustOutputBuffer( io ); + adjustInputBuffer( io ); } void -tr_peerIoWriteBuf( tr_peerIo * io, - struct evbuffer * buf ) +tr_peerIoSetBandwidthUnlimited( tr_peerIo * io, + tr_direction direction ) { - const size_t n = EVBUFFER_LENGTH( buf ); - tr_peerIoWrite( io, EVBUFFER_DATA(buf), n ); - evbuffer_drain( buf, n ); + struct tr_bandwidth * b; + + assert( io ); + assert( direction==TR_UP || direction==TR_DOWN ); + + b = io->bandwidth + direction; + b->isUnlimited = 1; + b->bytesUsed = 0; + b->bytesLeft = 0; + + adjustInputBuffer( io ); + adjustOutputBuffer( io ); +} + +double +tr_peerIoGetRateToClient( const tr_peerIo * io ) +{ + return tr_rcRate( io->speedometer[TR_DOWN] ); } +double +tr_peerIoGetRateToPeer( const tr_peerIo * io ) +{ + return tr_rcRate( io->speedometer[TR_UP] ); +} + + /** *** **/ @@ -455,7 +653,8 @@ tr_peerIoSetEncryption( tr_peerIo * io, int encryptionMode ) { assert( io ); - assert( encryptionMode==PEER_ENCRYPTION_NONE || encryptionMode==PEER_ENCRYPTION_RC4 ); + assert( encryptionMode==PEER_ENCRYPTION_NONE + || encryptionMode==PEER_ENCRYPTION_RC4 ); io->encryptionMode = encryptionMode; } @@ -470,6 +669,51 @@ tr_peerIoIsEncrypted( const tr_peerIo * *** **/ +int +tr_peerIoWantsBandwidth( const tr_peerIo * io, tr_direction direction ) +{ + assert( direction==TR_UP || direction==TR_DOWN ); + + if( direction == TR_DOWN ) + { + return TRUE; /* FIXME -- is there a good way to test for this? */ + } + else + { + return EVBUFFER_LENGTH( EVBUFFER_OUTPUT( io->bufev ) ) + || EVBUFFER_LENGTH( io->output ); + } +} + +void +tr_peerIoWrite( tr_peerIo * io, + const void * writeme, + size_t writemeLen ) +{ + assert( tr_amInEventThread( io->session ) ); + dbgmsg( io, "adding %zu bytes into io->output", writemeLen ); + + if( io->bandwidth[TR_UP].isUnlimited ) + bufferevent_write( io->bufev, writeme, writemeLen ); + else + evbuffer_add( io->output, writeme, writemeLen ); + + adjustOutputBuffer( io ); +} + +void +tr_peerIoWriteBuf( tr_peerIo * io, + struct evbuffer * buf ) +{ + const size_t n = EVBUFFER_LENGTH( buf ); + tr_peerIoWrite( io, EVBUFFER_DATA( buf ), n ); + evbuffer_drain( buf, n ); +} + +/** +*** +**/ + void tr_peerIoWriteBytes( tr_peerIo * io, struct evbuffer * outbuf, Index: libtransmission/torrent.h =================================================================== --- libtransmission/torrent.h (revision 6743) +++ libtransmission/torrent.h (working copy) @@ -54,8 +54,6 @@ tr_torrent* tr_torrentFindFromHash( tr_h tr_torrent* tr_torrentFindFromHashString( tr_handle *, const char * ); tr_torrent* tr_torrentFindFromObfuscatedHash( tr_handle *, const uint8_t* ); -void tr_torrentGetRates( const tr_torrent *, float * toClient, float * toPeer ); - int tr_torrentAllowsPex( const tr_torrent * ); /* get the index of this piece's first block */ @@ -120,10 +118,11 @@ struct tr_torrent tr_handle * handle; tr_info info; + int uploadLimit; tr_speedlimit uploadLimitMode; + int downloadLimit; tr_speedlimit downloadLimitMode; - struct tr_ratecontrol * upload; - struct tr_ratecontrol * download; + struct tr_ratecontrol * swarmSpeed; int error; @@ -180,6 +179,9 @@ struct tr_torrent tr_torrent * next; int uniqueId; + + double rateHistory[2][BANDWIDTH_PULSES_PER_SECOND]; + }; #endif Index: libtransmission/peer-msgs.c =================================================================== --- libtransmission/peer-msgs.c (revision 6743) +++ libtransmission/peer-msgs.c (working copy) @@ -69,7 +69,6 @@ enum PEX_INTERVAL = (90 * 1000), /* msec between sendPex() calls */ PEER_PULSE_INTERVAL = (250), /* msec between peerPulse() calls */ - RATE_PULSE_INTERVAL = (250), /* msec between ratePulse() calls */ MAX_QUEUE_SIZE = (100), @@ -320,8 +319,6 @@ struct tr_peermsgs struct request_list clientAskedFor; struct request_list clientWillAskFor; - tr_timer * rateTimer; - tr_timer * pulseTimer; tr_timer * pexTimer; time_t clientSentPexAt; @@ -857,9 +854,6 @@ pumpRequestQueue( tr_peermsgs * msgs ) } static int -peerPulse( void * vmsgs ); - -static int requestQueueIsFull( const tr_peermsgs * msgs ) { const int req_max = msgs->maxActiveRequests; @@ -1250,7 +1244,6 @@ static void clientGotBytes( tr_peermsgs * msgs, uint32_t byteCount ) { msgs->info->pieceDataActivityDate = time( NULL ); - tr_rcTransferred( msgs->info->rcToClient, byteCount ); fireClientGotData( msgs, byteCount ); } @@ -1464,26 +1457,9 @@ static void peerGotBytes( tr_peermsgs * msgs, uint32_t byteCount ) { msgs->info->pieceDataActivityDate = time( NULL ); - tr_rcTransferred( msgs->info->rcToPeer, byteCount ); firePeerGotData( msgs, byteCount ); } -static size_t -getDownloadMax( const tr_peermsgs * msgs ) -{ - static const size_t maxval = ~0; - const tr_torrent * tor = msgs->torrent; - - if( tor->downloadLimitMode == TR_SPEEDLIMIT_GLOBAL ) - return tor->handle->useDownloadLimit - ? tr_rcBytesLeft( tor->handle->download ) : maxval; - - if( tor->downloadLimitMode == TR_SPEEDLIMIT_SINGLE ) - return tr_rcBytesLeft( tor->download ); - - return maxval; -} - static void decrementDownloadedCount( tr_peermsgs * msgs, uint32_t byteCount ) { @@ -1564,12 +1540,6 @@ clientGotBlock( tr_peermsgs return 0; } -static void -didWrite( struct bufferevent * evin UNUSED, void * vmsgs ) -{ - peerPulse( vmsgs ); -} - static ReadState canRead( struct bufferevent * evin, void * vmsgs ) { @@ -1584,9 +1554,7 @@ canRead( struct bufferevent * evin, void } else if( msgs->state == AWAITING_BT_PIECE ) { - const size_t downloadMax = getDownloadMax( msgs ); - const size_t n = MIN( inlen, downloadMax ); - ret = n ? readBtPiece( msgs, in, n ) : READ_DONE; + ret = inlen ? readBtPiece( msgs, in, inlen ) : READ_DONE; } else switch( msgs->state ) { @@ -1611,47 +1579,14 @@ sendKeepalive( tr_peermsgs * msgs ) *** **/ -static size_t -getUploadMax( const tr_peermsgs * msgs ) -{ - static const size_t maxval = ~0; - const tr_torrent * tor = msgs->torrent; - int speedLeft; - int bufLeft; - - if( tor->uploadLimitMode == TR_SPEEDLIMIT_GLOBAL ) - speedLeft = tor->handle->useUploadLimit ? tr_rcBytesLeft( tor->handle->upload ) : maxval; - else if( tor->uploadLimitMode == TR_SPEEDLIMIT_SINGLE ) - speedLeft = tr_rcBytesLeft( tor->upload ); - else - speedLeft = INT_MAX; - - /* this basically says the outbuf shouldn't have more than one block - * queued up in it... blocksize, +13 for the size of the BT protocol's - * block message overhead */ - bufLeft = tor->blockSize + 13 - tr_peerIoWriteBytesWaiting( msgs->io ); - return MIN( speedLeft, bufLeft ); -} - -static int -getMaxBlocksFromPeerSoon( const tr_peermsgs * peer ) -{ - const double seconds = 30; - const double bytesPerSecond = peer->info->rateToClient * 1024; - const double totalBytes = bytesPerSecond * seconds; - const int blockCount = totalBytes / peer->torrent->blockSize; - /*fprintf( stderr, "rate %f -- blockCount %d\n", peer->info->rateToClient, blockCount );*/ - return blockCount; -} - static int ratePulse( void * vpeer ) { tr_peermsgs * peer = vpeer; - peer->info->rateToClient = tr_rcRate( peer->info->rcToClient ); - peer->info->rateToPeer = tr_rcRate( peer->info->rcToPeer ); + const double rateToClient = tr_peerIoGetRateToClient( peer->io ); + const int estimatedBlocksInNext30Seconds = ( rateToClient * 30 * 1024 ) / peer->torrent->blockSize; peer->minActiveRequests = 4; - peer->maxActiveRequests = peer->minActiveRequests + getMaxBlocksFromPeerSoon( peer ); + peer->maxActiveRequests = peer->minActiveRequests + estimatedBlocksInNext30Seconds; return TRUE; } @@ -1669,8 +1604,10 @@ popNextRequest( tr_peermsgs * msgs, stru static int peerPulse( void * vmsgs ) { - const time_t now = time( NULL ); tr_peermsgs * msgs = vmsgs; + const time_t now = time( NULL ); + + ratePulse( msgs ); tr_peerIoTryRead( msgs->io ); pumpRequestQueue( msgs ); @@ -1678,7 +1615,7 @@ peerPulse( void * vmsgs ) if( msgs->sendingBlock ) { - const size_t uploadMax = getUploadMax( msgs ); + const size_t uploadMax = tr_peerIoGetWriteBufferSpace( msgs->io ); size_t len = EVBUFFER_LENGTH( msgs->outBlock ); const size_t outlen = MIN( len, uploadMax ); @@ -1751,6 +1688,13 @@ peerPulse( void * vmsgs ) return TRUE; /* loop forever */ } +void +tr_peerMsgsPulse( tr_peermsgs * msgs ) +{ + if( msgs != NULL ) + peerPulse( msgs ); +} + static void gotError( struct bufferevent * evbuf UNUSED, short what, void * vmsgs ) { @@ -1983,8 +1927,6 @@ tr_peerMsgsNew( struct tr_torrent * torr m->info->peerIsInterested = 0; m->info->have = tr_bitfieldNew( torrent->info.pieceCount ); m->state = AWAITING_BT_LENGTH; - m->pulseTimer = tr_timerNew( m->session, peerPulse, m, PEER_PULSE_INTERVAL ); - m->rateTimer = tr_timerNew( m->session, ratePulse, m, RATE_PULSE_INTERVAL ); m->pexTimer = tr_timerNew( m->session, pexPulse, m, PEX_INTERVAL ); m->outMessages = evbuffer_new( ); m->outMessagesBatchedAt = 0; @@ -2004,7 +1946,7 @@ tr_peerMsgsNew( struct tr_torrent * torr sendBitfield( m ); tr_peerIoSetTimeoutSecs( m->io, 150 ); /* timeout after N seconds of inactivity */ - tr_peerIoSetIOFuncs( m->io, canRead, didWrite, gotError, m ); + tr_peerIoSetIOFuncs( m->io, canRead, NULL, gotError, m ); ratePulse( m ); return m; @@ -2015,8 +1957,6 @@ tr_peerMsgsFree( tr_peermsgs* msgs ) { if( msgs ) { - tr_timerFree( &msgs->pulseTimer ); - tr_timerFree( &msgs->rateTimer ); tr_timerFree( &msgs->pexTimer ); tr_publisherFree( &msgs->publisher ); reqListClear( &msgs->clientWillAskFor ); Index: libtransmission/peer-io.h =================================================================== --- libtransmission/peer-io.h (revision 6743) +++ libtransmission/peer-io.h (working copy) @@ -29,20 +29,20 @@ typedef struct tr_peerIo tr_peerIo; **/ tr_peerIo* - tr_peerIoNewOutgoing( struct tr_handle * handle, + tr_peerIoNewOutgoing( struct tr_handle * session, const struct in_addr * addr, int port, const uint8_t * torrentHash ); tr_peerIo* - tr_peerIoNewIncoming( struct tr_handle * handle, + tr_peerIoNewIncoming( struct tr_handle * session, const struct in_addr * addr, uint16_t port, int socket ); void tr_peerIoFree ( tr_peerIo * io ); -tr_handle* tr_peerIoGetHandle( tr_peerIo * io ); +tr_session* tr_peerIoGetSession( tr_peerIo * io ); /** *** @@ -109,13 +109,13 @@ void tr_peerIoSetIOFuncs( tr_peerIo tr_net_error_cb errcb, void * user_data ); -size_t tr_peerIoWriteBytesWaiting( const tr_peerIo * io ); +int tr_peerIoWantsBandwidth( const tr_peerIo * io, tr_direction ); void tr_peerIoTryRead( tr_peerIo * io ); void tr_peerIoWrite( tr_peerIo * io, const void * writeme, - int writeme_len ); + size_t writemeLen ); void tr_peerIoWriteBuf( tr_peerIo * io, struct evbuffer * buf ); @@ -178,5 +178,28 @@ void tr_peerIoDrain ( tr_peerIo struct evbuffer * inbuf, size_t byteCount ); +/** +*** +**/ + +size_t tr_peerIoGetBandwidthUsed( const tr_peerIo * io, + tr_direction direction ); + +size_t tr_peerIoGetBandwidthLeft( const tr_peerIo * io, + tr_direction direction ); + +size_t tr_peerIoGetWriteBufferSpace( const tr_peerIo * io ); + +void tr_peerIoSetBandwidth ( tr_peerIo * io, + tr_direction direction, + size_t bytesLeft ); + +void tr_peerIoSetBandwidthUnlimited( tr_peerIo * io, + tr_direction direction ); + +double tr_peerIoGetRateToClient( const tr_peerIo * io ); + +double tr_peerIoGetRateToPeer( const tr_peerIo * io ); + #endif Index: libtransmission/peer-msgs.h =================================================================== --- libtransmission/peer-msgs.h (revision 6743) +++ libtransmission/peer-msgs.h (working copy) @@ -35,6 +35,8 @@ void tr_peerMsgsSetChoke( tr_pee void tr_peerMsgsHave( tr_peermsgs * msgs, uint32_t pieceIndex ); +void tr_peerMsgsPulse( tr_peermsgs * msgs ); + #if 0 void tr_peerMsgsCancel( tr_peermsgs * msgs, uint32_t pieceIndex, Index: libtransmission/peer-mgr.c =================================================================== --- libtransmission/peer-mgr.c (revision 6743) +++ libtransmission/peer-mgr.c (working copy) @@ -122,12 +122,17 @@ Torrent; struct tr_peerMgr { - tr_handle * handle; + uint8_t bandwidthPulseNumber; + tr_session * session; tr_ptrArray * torrents; /* Torrent */ tr_ptrArray * incomingHandshakes; /* tr_handshake */ + tr_timer * bandwidthTimer; + double rateHistory[2][BANDWIDTH_PULSES_PER_SECOND]; + double globalPoolHistory[2][BANDWIDTH_PULSES_PER_SECOND]; }; #define tordbg(t, fmt...) tr_deepLog( __FILE__, __LINE__, t->tor->info.name, ##fmt ) +#define dbgmsg(fmt...) tr_deepLog(__FILE__,__LINE__,NULL,##fmt) /** *** @@ -136,12 +141,12 @@ struct tr_peerMgr static void managerLock( const struct tr_peerMgr * manager ) { - tr_globalLock( manager->handle ); + tr_globalLock( manager->session ); } static void managerUnlock( const struct tr_peerMgr * manager ) { - tr_globalUnlock( manager->handle ); + tr_globalUnlock( manager->session ); } static void torrentLock( Torrent * torrent ) @@ -156,7 +161,7 @@ torrentUnlock( Torrent * torrent ) static int torrentIsLocked( const Torrent * t ) { - return tr_globalIsLocked( t->manager->handle ); + return tr_globalIsLocked( t->manager->session ); } /** @@ -285,8 +290,6 @@ peerConstructor( const struct in_addr * { tr_peer * p; p = tr_new0( tr_peer, 1 ); - p->rcToClient = tr_rcInit( ); - p->rcToPeer = tr_rcInit( ); memcpy( &p->in_addr, in_addr, sizeof(struct in_addr) ); return p; } @@ -321,8 +324,6 @@ peerDestructor( tr_peer * peer ) tr_bitfieldFree( peer->have ); tr_bitfieldFree( peer->blame ); - tr_rcClose( peer->rcToClient ); - tr_rcClose( peer->rcToPeer ); tr_free( peer->client ); tr_free( peer ); } @@ -445,13 +446,17 @@ tr_peerMgrGenerateAllowedSet( const uint return a; } +static int bandwidthPulse( void * vmgr ); + tr_peerMgr* -tr_peerMgrNew( tr_handle * handle ) +tr_peerMgrNew( tr_session * session ) { tr_peerMgr * m = tr_new0( tr_peerMgr, 1 ); - m->handle = handle; + m->session = session; m->torrents = tr_ptrArrayNew( ); m->incomingHandshakes = tr_ptrArrayNew( ); + m->bandwidthPulseNumber = -1; + m->bandwidthTimer = tr_timerNew( session, bandwidthPulse, m, 1000/BANDWIDTH_PULSES_PER_SECOND ); return m; } @@ -460,6 +465,8 @@ tr_peerMgrFree( tr_peerMgr * manager ) { managerLock( manager ); + tr_timerFree( &manager->bandwidthTimer ); + /* free the handshakes. Abort invokes handshakeDoneCB(), which removes * the item from manager->handshakes, so this is a little roundabout... */ while( !tr_ptrArrayEmpty( manager->incomingHandshakes ) ) @@ -775,7 +782,7 @@ static void refillSoon( Torrent * t ) { if( t->refillTimer == NULL ) - t->refillTimer = tr_timerNew( t->manager->handle, + t->refillTimer = tr_timerNew( t->manager->session, refillPulse, t, REFILL_PERIOD_MSEC ); } @@ -804,8 +811,6 @@ peerCallbackFunc( void * vpeer, void * v tr_torrent * tor = t->tor; tor->activityDate = now; tor->uploadedCur += e->length; - tr_rcTransferred( tor->upload, e->length ); - tr_rcTransferred( tor->handle->upload, e->length ); tr_statsAddUploaded( tor->handle, e->length ); if( peer ) { struct peer_atom * atom = getExistingAtom( t, &peer->in_addr ); @@ -826,8 +831,6 @@ peerCallbackFunc( void * vpeer, void * v * into the jurisdiction of the tracker." */ if( peer ) tor->downloadedCur += e->length; - tr_rcTransferred( tor->download, e->length ); - tr_rcTransferred( tor->handle->download, e->length ); tr_statsAddDownloaded( tor->handle, e->length ); if( peer ) { struct peer_atom * atom = getExistingAtom( t, &peer->in_addr ); @@ -1040,7 +1043,7 @@ tr_peerMgrAddIncoming( tr_peerMgr * { managerLock( manager ); - if( tr_sessionIsAddressBlocked( manager->handle, addr ) ) + if( tr_sessionIsAddressBlocked( manager->session, addr ) ) { tr_dbg( "Banned IP address \"%s\" tried to connect to us", inet_ntoa( *addr ) ); @@ -1055,10 +1058,10 @@ tr_peerMgrAddIncoming( tr_peerMgr * tr_peerIo * io; tr_handshake * handshake; - io = tr_peerIoNewIncoming( manager->handle, addr, port, socket ); + io = tr_peerIoNewIncoming( manager->session, addr, port, socket ); handshake = tr_handshakeNew( io, - manager->handle->encryptionMode, + manager->session->encryptionMode, myHandshakeDoneCB, manager ); @@ -1078,7 +1081,7 @@ tr_peerMgrAddPex( tr_peerMgr * manag managerLock( manager ); t = getExistingTorrent( manager, torrentHash ); - if( !tr_sessionIsAddressBlocked( t->manager->handle, &pex->in_addr ) ) + if( !tr_sessionIsAddressBlocked( t->manager->session, &pex->in_addr ) ) ensureAtomExists( t, &pex->in_addr, pex->port, pex->flags, from ); managerUnlock( manager ); @@ -1225,11 +1228,11 @@ tr_peerMgrStartTorrent( tr_peerMgr * { t->isRunning = 1; - t->reconnectTimer = tr_timerNew( t->manager->handle, + t->reconnectTimer = tr_timerNew( t->manager->session, reconnectPulse, t, RECONNECT_PERIOD_MSEC ); - t->rechokeTimer = tr_timerNew( t->manager->handle, + t->rechokeTimer = tr_timerNew( t->manager->session, rechokePulse, t, RECHOKE_PERIOD_MSEC ); @@ -1394,7 +1397,9 @@ tr_peerMgrTorrentStats( const tr_peerMgr int * setmeWebseedsSendingToUs, int * setmePeersSendingToUs, int * setmePeersGettingFromUs, - int * setmePeersFrom ) + int * setmePeersFrom, + double * setmeRateToClient, + double * setmeRateToPeers ) { int i, size; const Torrent * t; @@ -1412,6 +1417,8 @@ tr_peerMgrTorrentStats( const tr_peerMgr *setmePeersGettingFromUs = 0; *setmePeersSendingToUs = 0; *setmeWebseedsSendingToUs = 0; + *setmeRateToClient = 0; + *setmeRateToPeers = 0; for( i=0; iflags & ADDED_F_SEED_FLAG ) ++*setmeSeedsConnected; + + *setmeRateToClient += tr_peerIoGetRateToClient( peer->io ); + + *setmeRateToPeers += tr_peerIoGetRateToPeer( peer->io ); } webseeds = (const tr_webseed **) tr_ptrArrayPeek( t->webseeds, &size ); @@ -1448,6 +1459,22 @@ tr_peerMgrTorrentStats( const tr_peerMgr managerUnlock( manager ); } +double +tr_peerMgrGetRate( const tr_peerMgr * manager, + tr_direction direction ) +{ + int i; + double rate = 0; + + assert( manager != NULL ); + assert( direction==TR_UP || direction==TR_DOWN ); + + for( i=0; irateHistory[direction][i]; + + return rate / 1024.0; +} + float* tr_peerMgrWebSpeeds( const tr_peerMgr * manager, const uint8_t * torrentHash ) @@ -1504,8 +1531,8 @@ tr_peerMgrPeerStats( const tr_peerMgr * stat->from = atom->from; stat->progress = peer->progress; stat->isEncrypted = tr_peerIoIsEncrypted( peer->io ) ? 1 : 0; - stat->rateToPeer = peer->rateToPeer; - stat->rateToClient = peer->rateToClient; + stat->rateToPeer = tr_peerIoGetRateToPeer( peer->io ); + stat->rateToClient = tr_peerIoGetRateToClient( peer->io ); stat->peerIsChoked = peer->peerIsChoked; stat->peerIsInterested = peer->peerIsInterested; stat->clientIsChoked = peer->clientIsChoked; @@ -1541,8 +1568,10 @@ tr_peerMgrPeerStats( const tr_peerMgr * struct ChokeData { - unsigned int doUnchoke : 1; - unsigned int isInterested : 1; + unsigned int doUnchoke : 1; + unsigned int isInterested : 1; + double rateToClient; + double rateToPeer; tr_peer * peer; }; @@ -1562,9 +1591,9 @@ compareChoke( const void * va, const voi int diff = 0; if( diff == 0 ) /* prefer higher dl speeds */ - diff = -tr_compareDouble( a->peer->rateToClient, b->peer->rateToClient ); + diff = -tr_compareDouble( a->rateToClient, b->rateToClient ); if( diff == 0 ) /* prefer higher ul speeds */ - diff = -tr_compareDouble( a->peer->rateToPeer, b->peer->rateToPeer ); + diff = -tr_compareDouble( a->rateToPeer, b->rateToPeer ); if( diff == 0 ) /* prefer unchoked */ diff = (int)a->peer->peerIsChoked - (int)b->peer->peerIsChoked; @@ -1606,6 +1635,8 @@ rechoke( Torrent * t ) struct ChokeData * node = &choke[size++]; node->peer = peer; node->isInterested = peer->peerIsInterested; + node->rateToClient = tr_peerIoGetRateToClient( peer->io ); + node->rateToPeer = tr_peerIoGetRateToPeer( peer->io ); } } @@ -1861,7 +1892,7 @@ getPeerCandidates( Torrent * t, int * se } /* Don't connect to peers in our blocklist */ - if( tr_sessionIsAddressBlocked( t->manager->handle, &atom->addr ) ) + if( tr_sessionIsAddressBlocked( t->manager->session, &atom->addr ) ) continue; ret[retCount++] = atom; @@ -1934,7 +1965,7 @@ reconnectPulse( void * vtorrent ) tordbg( t, "Starting an OUTGOING connection with %s", tr_peerIoAddrStr( &atom->addr, atom->port ) ); - io = tr_peerIoNewOutgoing( mgr->handle, &atom->addr, atom->port, t->hash ); + io = tr_peerIoNewOutgoing( mgr->session, &atom->addr, atom->port, t->hash ); if( io == NULL ) { atom->myflags |= MYFLAG_UNREACHABLE; @@ -1942,7 +1973,7 @@ reconnectPulse( void * vtorrent ) else { tr_handshake * handshake = tr_handshakeNew( io, - mgr->handle->encryptionMode, + mgr->session->encryptionMode, myHandshakeDoneCB, mgr ); @@ -1964,3 +1995,260 @@ reconnectPulse( void * vtorrent ) torrentUnlock( t ); return TRUE; } + +/**** +***** +***** +***** +****/ + +static double +allocateHowMuch( double desiredAvgKB, + const double * history, + int pulseNumber ) +{ + int i; + int oldest = ( pulseNumber + 1 ) % BANDWIDTH_PULSES_PER_SECOND; + const double baseline = desiredAvgKB * 1024.0 / BANDWIDTH_PULSES_PER_SECOND; + const double min = baseline * 0.66; + const double max = baseline * 1.33; + double bytes; + + bytes = desiredAvgKB * 1024.0; + for( i=0; i= 0.0 ); + assert( welfareBytes >= 0.0 ); + assert( direction==TR_UP || direction==TR_DOWN ); + + for( i=0; iio, direction ) ) + candidates[candidateCount++] = peers[i]; + else + tr_peerIoSetBandwidth( peers[i]->io, direction, 0 ); + + for( i=0; iio, direction ); + +if( direction==TR_UP ) fprintf( stderr, "%d peers, %d candidates, %.2f used, %.2f==merit, %.2f==welfare\n", peerCount, candidateCount, bytesUsed/1024.0, meritBytes/1024.0, welfareBytes/1024.0 ); + + welfare = welfareBytes / candidateCount; + + for( i=0; iio, direction ) ) / bytesUsed + : ( meritBytes / candidateCount ); +if( direction==TR_UP ) fprintf( stderr, "giving peer #%d %.1f KiB\n", i, (merit+welfare)/1024.0 ); + tr_peerIoSetBandwidth( peer->io, direction, merit + welfare ); + } + + /* cleanup */ + tr_free( candidates ); +} + +static size_t +countHandshakeBandwidth( tr_ptrArray * handshakes, tr_direction direction ) +{ + int i; + size_t total = 0; + const int n = tr_ptrArraySize( handshakes ); + for( i=0; iio, direction ); + } + return total; +} + +static void +givePeersUnlimitedBandwidth( tr_ptrArray * peers, tr_direction direction ) +{ + int i; + const int n = tr_ptrArraySize( peers ); + for( i=0; iio, direction ); + } +} + +static void +pumpAllPeers( tr_peerMgr * mgr ) +{ + int i, j; + const int torrentCount = tr_ptrArraySize( mgr->torrents ); + for( i=0; itorrents, i ); + const int peerCount = tr_ptrArraySize( t->peers ); + for( j=0; jpeers, j ); + tr_peerMsgsPulse( peer->msgs ); + } + } +} + + +/** + * Allocate bandwidth for each peer connection. + * + * @param mgr the peer manager + * @param direction whether to allocate upload or download bandwidth + * @return the amount of directional bandwidth used since the last pulse. + */ +static double +allocateBandwidth( tr_peerMgr * mgr, + tr_direction direction ) +{ + int i; + tr_session * session = mgr->session; + const int torrentCount = tr_ptrArraySize( mgr->torrents ); + Torrent ** torrents = (Torrent **) tr_ptrArrayBase( mgr->torrents ); + tr_ptrArray * globalPool = tr_ptrArrayNew( ); + double allBytesUsed = 0; + size_t poolBytesUsed = 0; + + assert( mgr ); + assert( direction==TR_UP || direction==TR_DOWN ); + + /* before allocating bandwidth, pump the connected peers */ + pumpAllPeers( mgr ); + + for( i=0; ipeers, direction ); + const size_t handshakeBytesUsed = + countHandshakeBandwidth( t->outgoingHandshakes, direction ); + + /* add this torrent's bandwidth use to allBytesUsed */ + allBytesUsed += peerBytesUsed + handshakeBytesUsed; + + /* process the torrent's peers based on its speed mode */ + switch( tr_torrentGetSpeedMode( t->tor, direction ) ) + { + case TR_SPEEDLIMIT_UNLIMITED: + givePeersUnlimitedBandwidth( t->peers, direction ); + break; + + case TR_SPEEDLIMIT_SINGLE: + setPeerBandwidth( t->peers, + direction, + t->tor->rateHistory[direction], + mgr->bandwidthPulseNumber, + tr_torrentGetSpeedLimit( t->tor, direction ) ); + break; + + case TR_SPEEDLIMIT_GLOBAL: { + int i; + const int n = tr_ptrArraySize( t->peers ); + for( i=0; ipeers, i ) ); + poolBytesUsed += peerBytesUsed + handshakeBytesUsed; + break; + } + } + } + + /* add incoming handshakes to the global pool */ + i = countHandshakeBandwidth( mgr->incomingHandshakes, direction ); + allBytesUsed += i; + poolBytesUsed += i; + + mgr->globalPoolHistory[direction][mgr->bandwidthPulseNumber] + = poolBytesUsed; + + /* handle the global pool's connections */ + if( tr_sessionIsSpeedLimitEnabled( session, direction ) ) + { + setPeerBandwidth( globalPool, + direction, + mgr->globalPoolHistory[direction], + mgr->bandwidthPulseNumber, + tr_sessionGetSpeedLimit( session, direction ) ); + } + else /* no speed limit on the global pool... */ + { + const int n = tr_ptrArraySize( globalPool ); + for( i=0; iio, direction ); + } + } + + /* now that we've allocated bandwidth, pump all the connected peers */ + pumpAllPeers( mgr ); + + /* cleanup */ + tr_ptrArrayFree( globalPool, NULL ); + return allBytesUsed; +} + +static int +bandwidthPulse( void * vmgr ) +{ + tr_peerMgr * mgr = vmgr; + int i; + managerLock( mgr ); + + /* keep track of how far we are into the cycle */ + if( ++mgr->bandwidthPulseNumber == BANDWIDTH_PULSES_PER_SECOND ) + mgr->bandwidthPulseNumber = 0; + + /* allocate the upload and download bandwidth */ + for( i=0; i<2; ++i ) + mgr->rateHistory[i][mgr->bandwidthPulseNumber] = + allocateBandwidth( mgr, i ); + + managerUnlock( mgr ); + return TRUE; +} Index: libtransmission/peer-mgr.h =================================================================== --- libtransmission/peer-mgr.h (revision 6743) +++ libtransmission/peer-mgr.h (working copy) @@ -110,7 +110,12 @@ void tr_peerMgrTorrentStats( const tr_pe int * setmeWebseedsSendingToUs, int * setmePeersSendingToUs, int * setmePeersGettingFromUs, - int * setmePeersFrom ); /* <-- array of TR_PEER_FROM__MAX */ + int * setmePeersFrom, /* <-- array of TR_PEER_FROM__MAX */ + double * setmeRateToClient, + double * setmeRateToPeers ); + +double tr_peerMgrGetRate( const tr_peerMgr * manager, + tr_direction direction ); struct tr_peer_stat * tr_peerMgrPeerStats( const tr_peerMgr * manager, const uint8_t * torrentHash, Index: libtransmission/handshake.c =================================================================== --- libtransmission/handshake.c (revision 6743) +++ libtransmission/handshake.c (working copy) @@ -1025,13 +1025,16 @@ tr_handshakeNew( tr_peerIo * i { tr_handshake * handshake; + tr_peerIoSetBandwidthUnlimited( io, TR_UP ); + tr_peerIoSetBandwidthUnlimited( io, TR_DOWN ); + handshake = tr_new0( tr_handshake, 1 ); handshake->io = io; handshake->crypto = tr_peerIoGetCrypto( io ); handshake->encryptionMode = encryptionMode; handshake->doneCB = doneCB; handshake->doneUserData = doneUserData; - handshake->handle = tr_peerIoGetHandle( io ); + handshake->handle = tr_peerIoGetSession( io ); tr_peerIoSetTimeoutSecs( io, 15 ); tr_peerIoSetIOFuncs( handshake->io, canRead, NULL, gotError, handshake ); @@ -1052,6 +1055,15 @@ tr_handshakeNew( tr_peerIo * i return handshake; } +struct tr_peerIo* +tr_handshakeGetIO( tr_handshake * handshake ) +{ + assert( handshake ); + assert( handshake->io ); + + return handshake->io; +} + const struct in_addr * tr_handshakeGetAddr( const struct tr_handshake * handshake, uint16_t * port ) { Index: libtransmission/handshake.h =================================================================== --- libtransmission/handshake.h (revision 6743) +++ libtransmission/handshake.h (working copy) @@ -35,4 +35,6 @@ const struct in_addr * tr_handshakeGetAd void tr_handshakeAbort( tr_handshake * handshake ); +struct tr_peerIo* tr_handshakeGetIO( tr_handshake * handshake ); + #endif Index: libtransmission/ratecontrol.c =================================================================== --- libtransmission/ratecontrol.c (revision 6743) +++ libtransmission/ratecontrol.c (working copy) @@ -29,10 +29,9 @@ #include "ratecontrol.h" #include "utils.h" -#define GRANULARITY_MSEC 500 -#define SHORT_INTERVAL_MSEC 1000 -#define LONG_INTERVAL_MSEC 8000 -#define HISTORY_SIZE (LONG_INTERVAL_MSEC / GRANULARITY_MSEC) +#define INTERVAL_MSEC 1000 +#define GRANULARITY_MSEC 200 +#define HISTORY_SIZE (INTERVAL_MSEC / GRANULARITY_MSEC) struct tr_transfer { @@ -83,7 +82,7 @@ tr_rcInit( void ) void tr_rcClose( tr_ratecontrol * r ) { - tr_rcReset( r ); + memset( r, 0, sizeof( tr_ratecontrol ) ); tr_free( r ); } @@ -91,29 +90,13 @@ tr_rcClose( tr_ratecontrol * r ) **** ***/ -size_t -tr_rcBytesLeft( const tr_ratecontrol * r ) -{ - size_t bytes = 0; - - if( r ) - { - const float cur = rateForInterval( r, SHORT_INTERVAL_MSEC ); - const float max = r->limit; - const float kb = max>cur ? max-cur : 0; - bytes = (size_t)(kb * 1024); - } - - return bytes; -} - float tr_rcRate( const tr_ratecontrol * r ) { float ret = 0.0f; if( r ) - ret = rateForInterval( r, LONG_INTERVAL_MSEC ); + ret = rateForInterval( r, INTERVAL_MSEC ); return ret; } @@ -137,13 +120,6 @@ tr_rcTransferred( tr_ratecontrol * r, si } void -tr_rcReset( tr_ratecontrol * r ) -{ - r->newest = 0; - memset( r->transfers, 0, sizeof(struct tr_transfer) * HISTORY_SIZE ); -} - -void tr_rcSetLimit( tr_ratecontrol * r, int limit ) { r->limit = limit; Index: libtransmission/ratecontrol.h =================================================================== --- libtransmission/ratecontrol.h (revision 6743) +++ libtransmission/ratecontrol.h (working copy) @@ -30,10 +30,8 @@ typedef struct tr_ratecontrol tr_ratecon tr_ratecontrol * tr_rcInit( void ); void tr_rcSetLimit( tr_ratecontrol *, int ); int tr_rcGetLimit( const tr_ratecontrol * ); -size_t tr_rcBytesLeft( const tr_ratecontrol * ); void tr_rcTransferred( tr_ratecontrol *, size_t byteCount ); float tr_rcRate( const tr_ratecontrol * ); -void tr_rcReset( tr_ratecontrol * ); void tr_rcClose( tr_ratecontrol * ); #endif Index: libtransmission/peer-mgr-private.h =================================================================== --- libtransmission/peer-mgr-private.h (revision 6743) +++ libtransmission/peer-mgr-private.h (working copy) @@ -63,12 +63,6 @@ typedef struct tr_peer struct tr_peermsgs * msgs; tr_publisher_tag msgsTag; - - struct tr_ratecontrol * rcToClient; - struct tr_ratecontrol * rcToPeer; - - float rateToClient; - float rateToPeer; } tr_peer; Index: libtransmission/transmission.h =================================================================== --- libtransmission/transmission.h (revision 6743) +++ libtransmission/transmission.h (working copy) @@ -534,25 +534,30 @@ tr_port_forwarding tr_sessionGetPortForw int tr_sessionCountTorrents( const tr_handle * h ); +typedef enum +{ + TR_CLIENT_TO_PEER = 0, TR_UP = 0, + TR_PEER_TO_CLIENT = 1, TR_DOWN = 1 +} +tr_direction; + void tr_sessionSetSpeedLimitEnabled( tr_handle * session, - int up_or_down, + tr_direction direction, int isEnabled ); -enum { TR_UP, TR_DOWN }; - void tr_sessionGetSpeed( const tr_handle * session, float * overall_down_KiBs, float * overall_up_KiBs ); int tr_sessionIsSpeedLimitEnabled( const tr_handle * session, - int up_or_down ); + tr_direction direction ); void tr_sessionSetSpeedLimit( tr_handle * session, - int up_or_down, + tr_direction direction, int KiB_sec ); -int tr_sessionGetSpeedLimit( const tr_handle * session, - int up_or_down ); +int tr_sessionGetSpeedLimit( const tr_handle * session, + tr_direction direction ); void tr_sessionSetPeerLimit( tr_handle * handle, uint16_t maxGlobalPeers ); @@ -843,18 +848,18 @@ typedef enum tr_speedlimit; void tr_torrentSetSpeedMode( tr_torrent * tor, - int up_or_down, + tr_direction up_or_down, tr_speedlimit mode ); tr_speedlimit tr_torrentGetSpeedMode( const tr_torrent * tor, - int up_or_down); + tr_direction direction ); void tr_torrentSetSpeedLimit( tr_torrent * tor, - int up_or_down, + tr_direction up_or_down, int KiB_sec ); int tr_torrentGetSpeedLimit( const tr_torrent * tor, - int up_or_down ); + tr_direction direction ); /**** ***** Peer Limits @@ -1246,10 +1251,10 @@ typedef struct tr_stat float percentDone; /** Download speed in KiB/s */ - float rateDownload; + double rateDownload; /** Upload speed in KiB/s */ - float rateUpload; + double rateUpload; #define TR_ETA_NOT_AVAIL -1 #define TR_ETA_UNKNOWN -2 Index: libtransmission/session.c =================================================================== --- libtransmission/session.c (revision 6743) +++ libtransmission/session.c (working copy) @@ -254,12 +254,9 @@ tr_sessionInitFull( const char * /* Initialize rate and file descripts controls */ - h->upload = tr_rcInit(); - tr_rcSetLimit( h->upload, uploadLimit ); + h->uploadLimit = uploadLimit; h->useUploadLimit = useUploadLimit; - - h->download = tr_rcInit(); - tr_rcSetLimit( h->download, downloadLimit ); + h->downloadLimit = downloadLimit; h->useDownloadLimit = useDownloadLimit; tr_fdInit( globalPeerLimit ); @@ -419,37 +416,40 @@ tr_sessionGetPortForwarding( const tr_ha ***/ void -tr_sessionSetSpeedLimitEnabled( tr_handle * h, - int up_or_down, - int use_flag ) +tr_sessionSetSpeedLimitEnabled( tr_handle * h, + tr_direction direction, + int use_flag ) { - if( up_or_down == TR_UP ) + assert( h ); + assert( direction==TR_UP || direction==TR_DOWN ); + + if( direction == TR_UP ) h->useUploadLimit = use_flag ? 1 : 0; else h->useDownloadLimit = use_flag ? 1 : 0; } int -tr_sessionIsSpeedLimitEnabled( const tr_handle * h, int up_or_down ) +tr_sessionIsSpeedLimitEnabled( const tr_handle * h, tr_direction direction ) { - return up_or_down==TR_UP ? h->useUploadLimit : h->useDownloadLimit; + return direction==TR_UP ? h->useUploadLimit : h->useDownloadLimit; } void tr_sessionSetSpeedLimit( tr_handle * h, - int up_or_down, + tr_direction direction, int KiB_sec ) { - if( up_or_down == TR_DOWN ) - tr_rcSetLimit( h->download, KiB_sec ); + if( direction == TR_DOWN ) + h->downloadLimit = KiB_sec; else - tr_rcSetLimit( h->upload, KiB_sec ); + h->uploadLimit = KiB_sec; } int -tr_sessionGetSpeedLimit( const tr_handle * h, int up_or_down ) +tr_sessionGetSpeedLimit( const tr_handle * h, tr_direction direction ) { - return tr_rcGetLimit( up_or_down==TR_UP ? h->upload : h->download ); + return direction==TR_UP ? h->uploadLimit : h->downloadLimit; } /*** @@ -479,9 +479,10 @@ tr_sessionGetSpeed( const tr_handle * s float * toPeer ) { if( session && toClient ) - *toClient = tr_rcRate( session->download ); + *toClient = tr_peerMgrGetRate( session->peerMgr, TR_DOWN ); + if( session && toPeer ) - *toPeer = tr_rcRate( session->upload ); + *toPeer = tr_peerMgrGetRate( session->peerMgr, TR_UP ); } int @@ -531,9 +532,6 @@ tr_closeAllConnections( void * vsession tr_peerMgrFree( session->peerMgr ); - tr_rcClose( session->upload ); - tr_rcClose( session->download ); - tr_trackerSessionClose( session ); tr_list_free( &session->blocklists, (TrListForeachFunc)_tr_blocklistFree ); tr_webClose( &session->web ); Index: libtransmission/session.h =================================================================== --- libtransmission/session.h (revision 6743) +++ libtransmission/session.h (working copy) @@ -35,6 +35,11 @@ #endif #endif +/** + * How frequently to reallocate peer bandwidth. + */ +#define BANDWIDTH_PULSES_PER_SECOND 5 + typedef enum { TR_NET_OK, TR_NET_ERROR, TR_NET_WAIT } tr_tristate_t; uint8_t* tr_peerIdNew( void ); @@ -81,8 +86,8 @@ struct tr_handle char * proxyUsername; char * proxyPassword; - struct tr_ratecontrol * upload; - struct tr_ratecontrol * download; + int uploadLimit; + int downloadLimit; struct tr_list * blocklists; struct tr_peerMgr * peerMgr; Index: libtransmission/torrent.c =================================================================== --- libtransmission/torrent.c (revision 6743) +++ libtransmission/torrent.c (working copy) @@ -137,38 +137,45 @@ tr_torrentUnlock( const tr_torrent * tor void tr_torrentSetSpeedMode( tr_torrent * tor, - int up_or_down, + tr_direction direction, tr_speedlimit mode ) { - tr_speedlimit * limit = up_or_down==TR_UP - ? &tor->uploadLimitMode - : &tor->downloadLimitMode; + tr_speedlimit * limit = direction==TR_UP ? &tor->uploadLimitMode + : &tor->downloadLimitMode; *limit = mode; } tr_speedlimit -tr_torrentGetSpeedMode( const tr_torrent * tor, - int up_or_down) +tr_torrentGetSpeedMode( const tr_torrent * tor, + tr_direction direction ) { - return up_or_down==TR_UP ? tor->uploadLimitMode - : tor->downloadLimitMode; + return direction==TR_UP ? tor->uploadLimitMode + : tor->downloadLimitMode; } void tr_torrentSetSpeedLimit( tr_torrent * tor, - int up_or_down, + tr_direction direction, int single_KiB_sec ) { - tr_ratecontrol * rc = up_or_down==TR_UP ? tor->upload : tor->download; - tr_rcSetLimit( rc, single_KiB_sec ); + switch( direction ) + { + case TR_UP: tor->uploadLimit = single_KiB_sec; break; + case TR_DOWN: tor->downloadLimit = single_KiB_sec; break; + default: assert( 0 ); + } } int tr_torrentGetSpeedLimit( const tr_torrent * tor, - int up_or_down ) + tr_direction direction ) { - tr_ratecontrol * rc = up_or_down==TR_UP ? tor->upload : tor->download; - return tr_rcGetLimit( rc ); + switch( direction ) + { + case TR_UP: return tor->uploadLimit; + case TR_DOWN: return tor->downloadLimit; + default: assert( 0 ); + } } /*** @@ -463,9 +470,9 @@ torrentRealInit( tr_handle * h, tr_torrentInitFilePieces( tor ); - tor->upload = tr_rcInit(); - tor->download = tr_rcInit(); - tor->swarmSpeed = tr_rcInit(); + tor->uploadLimit = 0; + tor->downloadLimit = 0; + tor->swarmSpeed = tr_rcInit(); tr_sha1( tor->obfuscatedHash, "req2", 4, info->hash, SHA_DIGEST_LENGTH, @@ -649,22 +656,6 @@ tr_torrentCanManualUpdate( const tr_torr && ( tr_trackerCanManualAnnounce( tor->tracker ) ); } -/* rcRate's averaging code can make it appear that we're - * still sending bytes after a torrent stops or all the - * peers disconnect, so short-circuit that appearance here */ -void -tr_torrentGetRates( const tr_torrent * tor, - float * toClient, - float * toPeer) -{ - const int showSpeed = tor->isRunning - && tr_peerMgrHasConnections( tor->handle->peerMgr, tor->info.hash ); - - if( toClient ) - *toClient = showSpeed ? tr_rcRate( tor->download ) : 0.0; - if( toPeer ) - *toPeer = showSpeed ? tr_rcRate( tor->upload ) : 0.0; -} const tr_info * tr_torrentInfo( const tr_torrent * tor ) { @@ -735,7 +726,9 @@ tr_torrentStat( tr_torrent * tor ) &s->webseedsSendingToUs, &s->peersSendingToUs, &s->peersGettingFromUs, - s->peersFrom ); + s->peersFrom, + &s->rateDownload, + &s->rateUpload ); usableSeeds += tor->info.webseedCount; @@ -749,8 +742,6 @@ tr_torrentStat( tr_torrent * tor ) ? 1.0 - (tr_torrentCountUncheckedPieces( tor ) / (double) tor->info.pieceCount) : 0.0; - tr_torrentGetRates( tor, &s->rateDownload, &s->rateUpload ); - s->swarmSpeed = tr_rcRate( tor->swarmSpeed ); s->activityDate = tor->activityDate; @@ -972,8 +963,6 @@ freeTorrent( tr_torrent * tor ) tr_cpClose( tor->completion ); - tr_rcClose( tor->upload ); - tr_rcClose( tor->download ); tr_rcClose( tor->swarmSpeed ); tr_trackerUnsubscribe( tor->tracker, tor->trackerSubscription ); Index: gtk/details.c =================================================================== --- gtk/details.c (revision 6744) +++ gtk/details.c (working copy) @@ -1085,7 +1085,6 @@ options_page_new ( TrTorrent * gtor ) uint16_t maxConnectedPeers; int i, row; gboolean b; - GtkAdjustment *a; GtkWidget *t, *w, *tb; tr_torrent * tor = tr_torrent_handle (gtor);