rendered paste bodyIndex: libtransmission/peer-io.c===================================================================--- libtransmission/peer-io.c (revision 6743)+++ libtransmission/peer-io.c (working copy)@@ -11,6 +11,7 @@ */ #include <assert.h>+#include <limits.h> /* INT_MAX */ #include <string.h> #include <stdio.h> #include <unistd.h>@@ -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];+ }; #endifIndex: 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 );+ #endifIndex: 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; i<TR_PEER_FROM__MAX; ++i ) setmePeersFrom[i] = 0;@@ -1436,6 +1443,10 @@ tr_peerMgrTorrentStats( const tr_peerMgr if( atom->flags & 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; i<BANDWIDTH_PULSES_PER_SECOND; ++i )+ rate += manager->rateHistory[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<BANDWIDTH_PULSES_PER_SECOND; ++i )+ if( i != oldest )+ bytes -= history[i];++ /* clamp the return value to lessen any oscillation */+ bytes = MAX( bytes, min );+ bytes = MIN( bytes, max );+ return bytes;+}++/**+ * Given a fixed amount of bandwidth (meritBytes + welfareBytes),+ * distribute it to the specified set of peers.+ *+ * @param peerArray peers whose client-to-peer bandwidth will be adjusted+ * @param direction whether to allocate upload or download bandwidth+ * @param history recent bandwidth history for these peers+ * @param pulseNumber index of the current pulse in the history array+ * @param desiredAvgKB overall bandwidth goal for these peers+ */+static void+setPeerBandwidth( tr_ptrArray * peerArray,+ const tr_direction direction,+ const double * history,+ int pulseNumber,+ double desiredAvgKB )+{+ const int peerCount = tr_ptrArraySize( peerArray );+ tr_peer ** peers = (tr_peer**) tr_ptrArrayBase( peerArray );+ int i;+ int candidateCount = 0;+ double welfare;+ size_t bytesUsed = 0;+ tr_peer ** candidates = tr_new( tr_peer*, peerCount );+ const double bytesThisPulse = allocateHowMuch( desiredAvgKB, history, pulseNumber );+ const double welfareBytes = MIN( 2048, bytesThisPulse * 0.2 );+ const double meritBytes = MAX( 0, bytesThisPulse - welfare );++ assert( meritBytes >= 0.0 );+ assert( welfareBytes >= 0.0 );+ assert( direction==TR_UP || direction==TR_DOWN );++ for( i=0; i<peerCount; ++i )+ if( tr_peerIoWantsBandwidth( peers[i]->io, direction ) )+ candidates[candidateCount++] = peers[i];+ else+ tr_peerIoSetBandwidth( peers[i]->io, direction, 0 );++ for( i=0; i<candidateCount; ++i )+ bytesUsed += tr_peerIoGetBandwidthUsed( candidates[i]->io, 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; i<candidateCount; ++i )+ {+ tr_peer * peer = candidates[i];+ const double merit = bytesUsed+ ? ( meritBytes * tr_peerIoGetBandwidthUsed( peer->io, 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; i<n; ++i ) {+ tr_peerIo * io = tr_handshakeGetIO( tr_ptrArrayNth( handshakes, i ) );+ total += tr_peerIoGetBandwidthUsed( io, direction );+ }+ return total;+}++static size_t+countPeerBandwidth( tr_ptrArray * peers, tr_direction direction )+{+ int i;+ size_t total = 0;+ const int n = tr_ptrArraySize( peers );+ for( i=0; i<n; ++i ) {+ tr_peer * peer = tr_ptrArrayNth( peers, i );+ total += tr_peerIoGetBandwidthUsed( peer->io, direction );+ }+ return total;+}++static void+givePeersUnlimitedBandwidth( tr_ptrArray * peers, tr_direction direction )+{+ int i;+ const int n = tr_ptrArraySize( peers );+ for( i=0; i<n; ++i ) {+ tr_peer * peer = tr_ptrArrayNth( peers, i );+ tr_peerIoSetBandwidthUnlimited( peer->io, direction );+ }+}++static void+pumpAllPeers( tr_peerMgr * mgr )+{+ int i, j;+ const int torrentCount = tr_ptrArraySize( mgr->torrents );+ for( i=0; i<torrentCount; ++i )+ {+ Torrent * t = tr_ptrArrayNth( mgr->torrents, i );+ const int peerCount = tr_ptrArraySize( t->peers );+ for( j=0; j<peerCount; ++j ) {+ tr_peer * peer = tr_ptrArrayNth( t->peers, 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; i<torrentCount; ++i )+ {+ Torrent * t = torrents[i];+ const size_t peerBytesUsed = countPeerBandwidth( t->peers, 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; i<n; ++i )+ tr_ptrArrayAppend( globalPool, tr_ptrArrayNth( t->peers, 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; i<n; ++i ) {+ tr_peer * peer = tr_ptrArrayNth( globalPool, i );+ tr_peerIoSetBandwidthUnlimited( peer->io, 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 );+ #endifIndex: 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 * ); #endifIndex: 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 -2Index: 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);