Changeset 3295

Timestamp:

Oct 6, 2007, 6:20:52 PM (15 years ago)

Author:

charles

Message:

• add a per-peer request queue to hold the next 10-15 seconds' worth of requests so that we always have more requests at hand when the current requests start to run low.
• increase the tracker `numwant' variable to grow our peer pool
• bugfixes in cancelling requests.
• make the debug log sexy and readable like uTorrent's ;)

Location:

trunk/libtransmission

Files:

• handshake.c (modified) (3 diffs)
• peer-mgr-private.h (modified) (1 diff)
• peer-mgr.c (modified) (17 diffs)
• peer-msgs.c (modified) (44 diffs)
• peer-msgs.h (modified) (1 diff)
• tracker.c (modified) (1 diff)
• transmission.c (modified) (1 diff)
• utils.c (modified) (1 diff)
• utils.h (modified) (1 diff)

trunk/libtransmission/handshake.c

@@ -158 +158 @@
     {
         va_list args;
-        const char * addr = tr_peerIoGetAddrStr( handshake->io );
         struct evbuffer * buf = evbuffer_new( );
-        evbuffer_add_printf( buf, "[%s:%d] %s (%p) ", file, line, addr, handshake->io );
+        char timestr[64];
+        evbuffer_add_printf( buf, "[%s] %s: ",
+                             tr_getLogTimeStr( timestr, sizeof(timestr) ),
+                             tr_peerIoGetAddrStr( handshake->io ) );
         va_start( args, fmt );
         evbuffer_add_vprintf( buf, fmt, args );
         va_end( args );
-        fprintf( stderr, "%s\n", EVBUFFER_DATA(buf) );
+        evbuffer_add_printf( buf, " (%s:%d)\n", file, line );
+
+        fwrite( EVBUFFER_DATA(buf), 1, EVBUFFER_LENGTH(buf), fp );
         evbuffer_free( buf );
     }
@@ -581 +585 @@
 /* FIXME: use  readHandshake here */
 
-    dbgmsg( handshake, "payload: need %d, got %d\n", (int)HANDSHAKE_SIZE, (int)EVBUFFER_LENGTH(inbuf) );
+    dbgmsg( handshake, "payload: need %d, got %d", (int)HANDSHAKE_SIZE, (int)EVBUFFER_LENGTH(inbuf) );
 
     if( EVBUFFER_LENGTH(inbuf) < HANDSHAKE_SIZE )
@@ -975 +979 @@
     struct evbuffer * inbuf = EVBUFFER_INPUT ( evin );
     ReadState ret;
-    dbgmsg( handshake, "handling canRead; state is [%s]\n", getStateName(handshake->state) );
+    dbgmsg( handshake, "handling canRead; state is [%s]", getStateName(handshake->state) );
 
     switch( handshake->state )

trunk/libtransmission/peer-mgr-private.h

@@ -57 +57 @@
     tr_publisher_tag msgsTag;
 
-    struct tr_ratecontrol * rateToClient;
-    struct tr_ratecontrol * rateToPeer;
+    struct tr_ratecontrol * rcToClient;
+    struct tr_ratecontrol * rcToPeer;
+
+    double rateToClient;
+    double rateToPeer;
 }
 tr_peer;

trunk/libtransmission/peer-mgr.c

@@ -40 +40 @@
 {
     /* how frequently to change which peers are choked */
-    RECHOKE_PERIOD_MSEC = (15 * 1000),
+    RECHOKE_PERIOD_MSEC = (1000),
 
     /* how frequently to decide which peers live and die */
@@ -46 +46 @@
 
     /* how frequently to refill peers' request lists */
-    REFILL_PERIOD_MSEC = 1500,
+    REFILL_PERIOD_MSEC = 666,
 
     /* don't change a peer's choke status more often than this */
@@ -61 +61 @@
     /* how many peers to unchoke per-torrent. */
     /* FIXME: make this user-configurable? */
-    NUM_UNCHOKED_PEERS_PER_TORRENT = 16, /* arbitrary */
-
-    /* another arbitrary number */
-    MAX_RECONNECTIONS_PER_MINUTE = 60,
-
-    MAX_RECONNECTIONS_PER_PULSE =
-        ((MAX_RECONNECTIONS_PER_MINUTE * RECONNECT_PERIOD_MSEC) / (60*1000)),
+    NUM_UNCHOKED_PEERS_PER_TORRENT = 12, /* arbitrary */
+
+    /* set this too high and there will be a lot of churn.
+     * set it too low and you'll get peers too slowly */
+    MAX_RECONNECTIONS_PER_PULSE = 8,
 
     /* corresponds to ut_pex's added.f flags */
@@ -132 +130 @@
         va_list args;
         struct evbuffer * buf = evbuffer_new( );
-        evbuffer_add_printf( buf, "[%s:%d] %s ", file, line, t->tor->info.name  );
+        char timestr[64];
+        evbuffer_add_printf( buf, "[%s] %s: ",
+                             tr_getLogTimeStr( timestr, sizeof(timestr) ),
+                             t->tor->info.name );
         va_start( args, fmt );
         evbuffer_add_vprintf( buf, fmt, args );
         va_end( args );
-        fprintf( fp, "%s\n", EVBUFFER_DATA(buf) );
+        evbuffer_add_printf( buf, " (%s:%d)\n", file, line );
+
+        fwrite( EVBUFFER_DATA(buf), 1, EVBUFFER_LENGTH(buf), fp );
         evbuffer_free( buf );
     }
@@ -302 +305 @@
     tr_peer * p;
     p = tr_new0( tr_peer, 1 );
-    p->rateToClient = tr_rcInit( );
-    p->rateToPeer = tr_rcInit( );
+    p->rcToClient = tr_rcInit( );
+    p->rcToPeer = tr_rcInit( );
     memcpy( &p->in_addr, in_addr, sizeof(struct in_addr) );
     return p;
@@ -339 +342 @@
     tr_bitfieldFree( peer->blame );
     tr_bitfieldFree( peer->banned );
-    tr_rcClose( peer->rateToClient );
-    tr_rcClose( peer->rateToPeer );
+    tr_rcClose( peer->rcToClient );
+    tr_rcClose( peer->rcToPeer );
     tr_free( peer->client );
     tr_free( peer );
@@ -698 +701 @@
 
     torrentLock( t );
+    tordbg( t, "Refilling Request Buffers..." );
 
     blocks = getPreferredBlocks( t, &blockCount );
@@ -747 +751 @@
 
     t->refillTimer = NULL;
-
     torrentUnlock( t );
     return FALSE;
@@ -796 +799 @@
                                               refillPulse, t,
                                               REFILL_PERIOD_MSEC );
+            break;
+
+        case TR_PEERMSG_CANCEL:
+            tr_bitfieldRem( t->requested, _tr_block( t->tor, e->pieceIndex, e->offset ) );
             break;
 
@@ -906 +913 @@
         const struct in_addr * addr = tr_peerIoGetAddress( io,  &port );
         ensureAtomExists( t, addr, port, 0, TR_PEER_FROM_INCOMING );
+        tr_free( peer->client );
+        peer->client = peer_id ? tr_clientForId( peer_id ) : NULL;
         if( peer->msgs != NULL ) { /* we already have this peer */
             tr_peerIoFree( io );
@@ -913 +922 @@
             peer->io = io;
             peer->msgs = tr_peerMsgsNew( t->tor, peer, msgsCallbackFunc, t, &peer->msgsTag );
-            tr_free( peer->client );
-            peer->client = peer_id ? tr_clientForId( peer_id ) : NULL;
         }
     }
@@ -1258 +1265 @@
         ++setmePeersFrom[atom->from];
 
-        if( tr_rcRate( peer->rateToPeer ) > 0.01 )
+        if( peer->rateToPeer > 0.01 )
             ++*setmePeersGettingFromUs;
 
-        if( tr_rcRate( peer->rateToClient ) > 0.01 )
+        if( peer->rateToClient > 0.01 )
             ++*setmePeersSendingToUs;
     }
@@ -1298 +1305 @@
         stat->progress         = peer->progress;
         stat->isEncrypted      = tr_peerIoIsEncrypted( peer->io ) ? 1 : 0;
-        stat->uploadToRate     = tr_rcRate( peer->rateToPeer );
-        stat->downloadFromRate = tr_rcRate( peer->rateToClient );
+        stat->uploadToRate     = peer->rateToPeer;
+        stat->downloadFromRate = peer->rateToClient;
         stat->isDownloading    = stat->uploadToRate > 0.01;
         stat->isUploading      = stat->downloadFromRate > 0.01;
@@ -1356 +1363 @@
 **/
 
+static double
+getWeightedThroughput( const tr_peer * peer )
+{
+    return ( 3 * peer->rateToPeer )
+         + ( 1 * peer->rateToClient );
+}
+
 static void
 rechoke( Torrent * t )
@@ -1378 +1392 @@
         node->preferred = peer->peerIsInterested && !clientIsSnubbedBy(peer);
         node->randomKey = tr_rand( INT_MAX );
-        node->rate = (3*tr_rcRate(peer->rateToPeer))
-                   + (1*tr_rcRate(peer->rateToClient));
+        node->rate = getWeightedThroughput( peer );
     }
 
@@ -1439 +1452 @@
         const struct peer_atom * atom = getExistingAtom( t, &peer->in_addr );
         const int peerIsSeed = atom->flags & ADDED_F_SEED_FLAG;
-        const double throughput = (3*tr_rcRate(peer->rateToPeer))
-                                + (1*tr_rcRate(peer->rateToClient));
+        const double throughput = getWeightedThroughput( peer );
 
         assert( atom != NULL );
@@ -1546 +1558 @@
         /* add some new ones */
         nAdd = MAX_CONNECTED_PEERS_PER_TORRENT - peerCount;
-        for( i=0; i<nAdd && i<nCandidates; ++i )
+        for( i=0; i<nAdd && i<nCandidates && i<MAX_RECONNECTIONS_PER_PULSE; ++i )
+        //for( i=0; i<nCandidates; ++i )
         {
             tr_peerMgr * mgr = t->manager;

trunk/libtransmission/peer-msgs.c

@@ -69 +69 @@
     KEEPALIVE_INTERVAL_SECS = 90,          /* idle seconds before we send a keepalive */
     PEX_INTERVAL            = (60 * 1000), /* msec between calls to sendPex() */
-    PEER_PULSE_INTERVAL     = (33),        /* msec between calls to pulse() */
+    PEER_PULSE_INTERVAL     = (66),        /* msec between calls to pulse() */
+    RATE_PULSE_INTERVAL     = (333),       /* msec between calls to ratePulse() */
 };
 
@@ -88 +89 @@
 
 static int
-peer_request_compare( const void * va, const void * vb )
+compareRequest( const void * va, const void * vb )
 {
     struct peer_request * a = (struct peer_request*) va;
@@ -112 +113 @@
     tr_list * peerAskedFor;
     tr_list * clientAskedFor;
-
+    tr_list * clientWillAskFor;
+
+    tr_timer * rateTimer;
     tr_timer * pulseTimer;
     tr_timer * pexTimer;
@@ -131 +134 @@
     
     uint8_t state;
-
     uint8_t ut_pex_id;
-
     uint16_t pexCount;
-
     uint32_t incomingMessageLength;
+    uint32_t maxActiveRequests;
+    uint32_t minActiveRequests;
 
     tr_pex * pex;
@@ -154 +156 @@
     {
         va_list args;
-        const char * addr = tr_peerIoGetAddrStr( msgs->io );
         struct evbuffer * buf = evbuffer_new( );
-        evbuffer_add_printf( buf, "[%s:%d] %s (%p) ", file, line, addr, msgs->io );
+        char timestr[64];
+        evbuffer_add_printf( buf, "[%s] %s [%s]: ",
+                             tr_getLogTimeStr( timestr, sizeof(timestr) ),
+                             tr_peerIoGetAddrStr( msgs->io ),
+                             msgs->info->client );
         va_start( args, fmt );
         evbuffer_add_vprintf( buf, fmt, args );
         va_end( args );
-        fprintf( fp, "%s\n", EVBUFFER_DATA(buf) );
+        evbuffer_add_printf( buf, " (%s:%d)\n", file, line );
+
+        fwrite( EVBUFFER_DATA(buf), 1, EVBUFFER_LENGTH(buf), fp );
         evbuffer_free( buf );
     }
 }
 
-#define dbgmsg(handshake, fmt...) myDebug(__FILE__, __LINE__, handshake, ##fmt )
+#define dbgmsg(msgs, fmt...) myDebug(__FILE__, __LINE__, msgs, ##fmt )
+
+/**
+***
+**/
+
+static void
+protocolSendRequest( tr_peermsgs * msgs, const struct peer_request * req )
+{
+    tr_peerIo * io = msgs->io;
+    struct evbuffer * out = msgs->outMessages;
+
+    dbgmsg( msgs, "requesting %u:%u->%u", req->index, req->offset, req->length );
+    tr_peerIoWriteUint32( io, out, sizeof(uint8_t) + 3*sizeof(uint32_t) );
+    tr_peerIoWriteUint8 ( io, out, BT_REQUEST );
+    tr_peerIoWriteUint32( io, out, req->index );
+    tr_peerIoWriteUint32( io, out, req->offset );
+    tr_peerIoWriteUint32( io, out, req->length );
+}
+
+static void
+protocolSendCancel( tr_peermsgs * msgs, const struct peer_request * req )
+{
+    tr_peerIo * io = msgs->io;
+    struct evbuffer * out = msgs->outMessages;
+
+    dbgmsg( msgs, "cancelling %u:%u->%u", req->index, req->offset, req->length );
+    tr_peerIoWriteUint32( io, out, sizeof(uint8_t) + 3*sizeof(uint32_t) );
+    tr_peerIoWriteUint8 ( io, out, BT_CANCEL );
+    tr_peerIoWriteUint32( io, out, req->index );
+    tr_peerIoWriteUint32( io, out, req->offset );
+    tr_peerIoWriteUint32( io, out, req->length );
+}
+
+static void
+protocolSendHave( tr_peermsgs * msgs, uint32_t index )
+{
+    tr_peerIo * io = msgs->io;
+    struct evbuffer * out = msgs->outMessages;
+
+    dbgmsg( msgs, "sending Have %u", index );
+    tr_peerIoWriteUint32( io, out, sizeof(uint8_t) + sizeof(uint32_t) );
+    tr_peerIoWriteUint8 ( io, out, BT_HAVE );
+    tr_peerIoWriteUint32( io, out, index );
+}
+
+static void
+protocolSendChoke( tr_peermsgs * msgs, int choke )
+{
+    tr_peerIo * io = msgs->io;
+    struct evbuffer * out = msgs->outMessages;
+
+    dbgmsg( msgs, "sending %s", (choke ? "Choke" : "Unchoke") );
+    tr_peerIoWriteUint32( io, out, sizeof(uint8_t) );
+    tr_peerIoWriteUint8 ( io, out, choke ? BT_CHOKE : BT_UNCHOKE );
+}
 
 /**
@@ -192 +254 @@
     tr_peermsgs_event e = blankEvent;
     e.eventType = TR_PEERMSG_NEED_REQ;
+    dbgmsg( msgs, "firing NEED_REQ" );
     publish( msgs, &e );
 }
@@ -221 +284 @@
     e.offset = offset;
     e.length = length;
+    publish( msgs, &e );
+}
+
+static void
+fireCancelledReq( tr_peermsgs * msgs, const struct peer_request * req )
+{
+    tr_peermsgs_event e = blankEvent;
+    e.eventType = TR_PEERMSG_CANCEL;
+    e.pieceIndex = req->index;
+    e.offset = req->offset;
+    e.length = req->length;
     publish( msgs, &e );
 }
@@ -278 +352 @@
 
     msgs->info->clientIsInterested = weAreInterested;
-    dbgmsg( msgs, ": sending an %s message",
-            weAreInterested ? "INTERESTED" : "NOT_INTERESTED");
+    dbgmsg( msgs, "Sending %s",
+            weAreInterested ? "Interested" : "Not Interested");
 
     tr_peerIoWriteUint32( msgs->io, msgs->outMessages, sizeof(uint8_t) );
@@ -322 +396 @@
             }
 
-        dbgmsg( msgs, "sending a %s message", (choke ? "CHOKE" : "UNCHOKE") );
-        tr_peerIoWriteUint32( msgs->io, msgs->outMessages, sizeof(uint8_t) );
-        tr_peerIoWriteUint8 ( msgs->io, msgs->outMessages,
-                                             choke ? BT_CHOKE : BT_UNCHOKE );
+        protocolSendChoke( msgs, choke );
         msgs->info->chokeChangedAt = time( NULL );
     }
@@ -335 +406 @@
 
 void
-tr_peerMsgsCancel( tr_peermsgs * msgs,
-                   uint32_t      pieceIndex,
-                   uint32_t      offset,
-                   uint32_t      length )
-{
-    tr_list * node;
-    struct peer_request tmp;
-
-    assert( msgs != NULL );
-    assert( length > 0 );
-
-    /* have we asked the peer for this piece? */
-    tmp.index = pieceIndex;
-    tmp.offset = offset;
-    tmp.length = length;
-    node = tr_list_remove( &msgs->clientAskedFor, &tmp, peer_request_compare );
-
-    /* if so, send a cancel message */
-    if( node != NULL ) {
-        tr_peerIoWriteUint32( msgs->io, msgs->outMessages, sizeof(uint8_t) + 3*sizeof(uint32_t) );
-        tr_peerIoWriteUint8 ( msgs->io, msgs->outMessages, BT_CANCEL );
-        tr_peerIoWriteUint32( msgs->io, msgs->outMessages, pieceIndex );
-        tr_peerIoWriteUint32( msgs->io, msgs->outMessages, offset );
-        tr_peerIoWriteUint32( msgs->io, msgs->outMessages, length );
-        tr_free( node );
-    }
-}
-
-/**
-***
-**/
-
-void
 tr_peerMsgsHave( tr_peermsgs * msgs,
-                 uint32_t      pieceIndex )
-{
-    dbgmsg( msgs, "w00t telling them we HAVE piece #%d", pieceIndex );
-
-    tr_peerIoWriteUint32( msgs->io, msgs->outMessages,
-                               sizeof(uint8_t) + sizeof(uint32_t) );
-    tr_peerIoWriteUint8 ( msgs->io, msgs->outMessages, BT_HAVE );
-    tr_peerIoWriteUint32( msgs->io, msgs->outMessages, pieceIndex );
-
+                 uint32_t      index )
+{
+    protocolSendHave( msgs, index );
+
+    /* since we have more pieces now, we might not be interested in this peer */
     updateInterest( msgs );
 }
@@ -452 +486 @@
 
 static int
-pulse( void * vmsgs );
-
-static int
 reqIsValid( const tr_peermsgs * msgs, uint32_t index, uint32_t offset, uint32_t length )
 {
@@ -477 +508 @@
 }
 
+static void
+pumpRequestQueue( tr_peermsgs * msgs )
+{
+    const int max = msgs->maxActiveRequests;
+    const int min = msgs->minActiveRequests;
+    int count = tr_list_size( msgs->clientAskedFor );
+    int sent = 0;
+
+    if( count > min )
+        return;
+    if( msgs->info->clientIsChoked )
+        return;
+
+    while( ( count < max ) && ( msgs->clientWillAskFor != NULL ) )
+    {
+        struct peer_request * req = tr_list_pop_front( &msgs->clientWillAskFor );
+        protocolSendRequest( msgs, req );
+        req->time_requested = msgs->lastReqAddedAt = time( NULL );
+        tr_list_append( &msgs->clientAskedFor, req );
+        ++count;
+        ++sent;
+    }
+
+    if( sent )
+        dbgmsg( msgs, "pump sent %d requests, now have %d active and %d queued",
+                sent,
+                tr_list_size(msgs->clientAskedFor),
+                tr_list_size(msgs->clientWillAskFor) );
+
+    if( count < max )
+        fireNeedReq( msgs );
+}
+
 int
 tr_peerMsgsAddRequest( tr_peermsgs * msgs,
@@ -483 +547 @@
                        uint32_t      length )
 {
+    const int req_max = msgs->maxActiveRequests;
     struct peer_request tmp, *req;
-    int maxSize;
 
     assert( msgs != NULL );
@@ -490 +554 @@
     assert( reqIsValid( msgs, index, offset, length ) );
 
-    if( msgs->info->clientIsChoked )
+    /**
+    ***  Reasons to decline the request
+    **/
+
+    /* don't send requests to choked clients */
+    if( msgs->info->clientIsChoked ) {
+        dbgmsg( msgs, "declining request because they're choking us" );
         return TR_ADDREQ_CLIENT_CHOKED;
-
+    }
+
+    /* peer doesn't have this piece */
     if( !tr_bitfieldHas( msgs->info->have, index ) )
         return TR_ADDREQ_MISSING;
 
-    maxSize = MIN( 3 + (int)(tr_rcRate(msgs->info->rateToClient)/5), 100 );
-    if( tr_list_size( msgs->clientAskedFor) >= maxSize )
+    /* peer's queue is full */
+    if( tr_list_size( msgs->clientWillAskFor ) >= req_max ) {
+        dbgmsg( msgs, "declining request because we're full" );
         return TR_ADDREQ_FULL;
+    }
 
     /* have we already asked for this piece? */
@@ -504 +578 @@
     tmp.offset = offset;
     tmp.length = length;
-    if( tr_list_remove( &msgs->clientAskedFor, &tmp, peer_request_compare ) != NULL )
+    if( tr_list_find( msgs->clientAskedFor, &tmp, compareRequest ) ) {
+        dbgmsg( msgs, "declining because it's a duplicate" );
         return TR_ADDREQ_DUPLICATE;
-
-    dbgmsg( msgs, "w00t peer has a max request queue size of %d... adding request for piece %d, offset %d", maxSize, (int)index, (int)offset );
-
-    /* queue the request */
-    tr_peerIoWriteUint32( msgs->io, msgs->outMessages, sizeof(uint8_t) + 3*sizeof(uint32_t) );
-    tr_peerIoWriteUint8 ( msgs->io, msgs->outMessages, BT_REQUEST );
-    tr_peerIoWriteUint32( msgs->io, msgs->outMessages, index );
-    tr_peerIoWriteUint32( msgs->io, msgs->outMessages, offset );
-    tr_peerIoWriteUint32( msgs->io, msgs->outMessages, length );
-
-    /* add it to our `requests sent' list */
-    req = tr_new( struct peer_request, 1 );
+    }
+    if( tr_list_find( msgs->clientWillAskFor, &tmp, compareRequest ) ) {
+        dbgmsg( msgs, "declining because it's a duplicate" );
+        return TR_ADDREQ_DUPLICATE;
+    }
+
+    /**
+    ***  Accept this request
+    **/
+
+    dbgmsg( msgs, "added req for piece %d, offset %d", (int)index, (int)offset );
+    req = tr_new0( struct peer_request, 1 );
     *req = tmp;
-    req->time_requested = msgs->lastReqAddedAt = time( NULL );
-    tr_list_append( &msgs->clientAskedFor, req );
-    pulse( msgs );
-
+    tr_list_append( &msgs->clientWillAskFor, req );
     return TR_ADDREQ_OK;
+}
+
+static void
+tr_peerMsgsCancelAllRequests( tr_peermsgs * msgs )
+{
+    struct peer_request * req;
+
+    while(( req = tr_list_pop_front( &msgs->clientWillAskFor ) ))
+    {
+        fireCancelledReq( msgs, req );
+        tr_free( req );
+    }
+
+    while(( req = tr_list_pop_front( &msgs->clientAskedFor ) ))
+    {
+        fireCancelledReq( msgs, req );
+        protocolSendCancel( msgs, req );
+        tr_free( req );
+    }
+}
+
+void
+tr_peerMsgsCancel( tr_peermsgs * msgs,
+                   uint32_t      pieceIndex,
+                   uint32_t      offset,
+                   uint32_t      length )
+{
+    struct peer_request *req, tmp;
+
+    assert( msgs != NULL );
+    assert( length > 0 );
+
+    /* have we asked the peer for this piece? */
+    tmp.index = pieceIndex;
+    tmp.offset = offset;
+    tmp.length = length;
+
+    /* if it's only in the queue and hasn't been sent yet, free it */
+    if(( req = tr_list_remove( &msgs->clientWillAskFor, &tmp, compareRequest ) ))
+    {
+        fireCancelledReq( msgs, req );
+        tr_free( req );
+    }
+
+    /* if it's already been sent, send a cancel message too */
+    if(( req = tr_list_remove( &msgs->clientAskedFor, &tmp, compareRequest ) ))
+    {
+        protocolSendCancel( msgs, req );
+        fireCancelledReq( msgs, req );
+        tr_free( req );
+    }
 }
 
@@ -712 +835 @@
 
     if( len == 0 ) /* peer sent us a keepalive message */
-        dbgmsg( msgs, "peer sent us a keepalive message..." );
+        dbgmsg( msgs, "got KeepAlive" );
     else {
-        dbgmsg( msgs, "peer is sending us a message with %"PRIu64" bytes...", (uint64_t)len );
         msgs->incomingMessageLength = len;
         msgs->state = AWAITING_BT_MESSAGE;
@@ -743 +865 @@
     tr_peerIoReadUint8( msgs->io, inbuf, &id );
     msglen--;
-    dbgmsg( msgs, "peer sent us a message... "
-                  "bt id number is %d, and remaining len is %d", (int)id, (int)msglen );
+    dbgmsg( msgs, "got BT id %d, len %d", (int)id, (int)msglen );
 
     switch( id )
     {
         case BT_CHOKE:
-            dbgmsg( msgs, "w00t peer sent us a BT_CHOKE" );
+            dbgmsg( msgs, "got Choke" );
             assert( msglen == 0 );
             msgs->info->clientIsChoked = 1;
-            
+#if 0
             tr_list * walk;
             for( walk = msgs->peerAskedFor; walk != NULL; )
@@ -766 +887 @@
                 walk = next;
             }
-            tr_list_free( &msgs->clientAskedFor, tr_free );
+#endif
+            tr_peerMsgsCancelAllRequests( msgs );
             break;
 
         case BT_UNCHOKE:
-            dbgmsg( msgs, "w00t peer sent us a BT_UNCHOKE" );
+            dbgmsg( msgs, "got Unchoke" );
             assert( msglen == 0 );
             msgs->info->clientIsChoked = 0;
@@ -777 +899 @@
 
         case BT_INTERESTED:
-            dbgmsg( msgs, "w00t peer sent us a BT_INTERESTED" );
+            dbgmsg( msgs, "got Interested" );
             assert( msglen == 0 );
             msgs->info->peerIsInterested = 1;
@@ -784 +906 @@
 
         case BT_NOT_INTERESTED:
-            dbgmsg( msgs, "w00t peer sent us a BT_NOT_INTERESTED" );
+            dbgmsg( msgs, "got Not Interested" );
             assert( msglen == 0 );
             msgs->info->peerIsInterested = 0;
@@ -790 +912 @@
 
         case BT_HAVE:
-            dbgmsg( msgs, "w00t peer sent us a BT_HAVE" );
             assert( msglen == 4 );
             tr_peerIoReadUint32( msgs->io, inbuf, &ui32 );
             tr_bitfieldAdd( msgs->info->have, ui32 );
             updatePeerProgress( msgs );
+            dbgmsg( msgs, "got Have: %u", ui32 );
             break;
 
         case BT_BITFIELD: {
             const int clientIsSeed = tr_cpGetStatus( msgs->torrent->completion ) != TR_CP_INCOMPLETE;
-            dbgmsg( msgs, "w00t peer sent us a BT_BITFIELD" );
             assert( msglen == msgs->info->have->len );
             tr_peerIoReadBytes( msgs->io, inbuf, msgs->info->have->bits, msglen );
@@ -810 +931 @@
         case BT_REQUEST: {
             struct peer_request * req;
-            dbgmsg( msgs, "peer sent us a BT_REQUEST" );
             assert( msglen == 12 );
             req = tr_new( struct peer_request, 1 );
@@ -816 +936 @@
             tr_peerIoReadUint32( msgs->io, inbuf, &req->offset );
             tr_peerIoReadUint32( msgs->io, inbuf, &req->length );
+            dbgmsg( msgs, "got Request: %u:%u->%u", req->index, req->offset, req->length );
             
             if ( !requestIsValid( msgs, req ) )
@@ -868 +989 @@
             struct peer_request req;
             void * data;
-            dbgmsg( msgs, "peer sent us a BT_CANCEL" );
             assert( msglen == 12 );
             tr_peerIoReadUint32( msgs->io, inbuf, &req.index );
             tr_peerIoReadUint32( msgs->io, inbuf, &req.offset );
             tr_peerIoReadUint32( msgs->io, inbuf, &req.length );
-            data = tr_list_remove( &msgs->peerAskedFor, &req, peer_request_compare );
+            dbgmsg( msgs, "got a Cancel %u:%u->%u", req.index, req.offset, req.length );
+            data = tr_list_remove( &msgs->peerAskedFor, &req, compareRequest );
             tr_free( data );
             break;
@@ -879 +1000 @@
 
         case BT_PIECE: {
-            dbgmsg( msgs, "peer sent us a BT_PIECE" );
+            dbgmsg( msgs, "got a Piece!" );
             assert( msgs->blockToUs.length == 0 );
             tr_peerIoReadUint32( msgs->io, inbuf, &msgs->blockToUs.index );
@@ -892 +1013 @@
 
         case BT_PORT: {
-            dbgmsg( msgs, "peer sent us a BT_PORT" );
+            dbgmsg( msgs, "Got a BT_PORT" );
             assert( msglen == 2 );
             tr_peerIoReadUint16( msgs->io, inbuf, &msgs->info->port );
@@ -905 +1026 @@
         case BT_HAVE_ALL: {
             assert( msglen == 0 );
-            dbgmsg( msgs, "peer sent us a BT_HAVE_ALL" );
+            dbgmsg( msgs, "Got a BT_HAVE_ALL" );
             memset( msgs->info->have->bits, 1, msgs->info->have->len );
             updatePeerProgress( msgs );
@@ -913 +1034 @@
         case BT_HAVE_NONE: {
             assert( msglen == 0 );
-            dbgmsg( msgs, "peer sent us a BT_HAVE_NONE" );
+            dbgmsg( msgs, "Got a BT_HAVE_NONE" );
             memset( msgs->info->have->bits, 1, msgs->info->have->len );
             updatePeerProgress( msgs );
@@ -923 +1044 @@
             tr_list * node;
             assert( msglen == 12 );
-            dbgmsg( msgs, "peer sent us a BT_REJECT" );
+            dbgmsg( msgs, "Got a BT_REJECT" );
             tr_peerIoReadUint32( msgs->io, inbuf, &req.index );
             tr_peerIoReadUint32( msgs->io, inbuf, &req.offset );
             tr_peerIoReadUint32( msgs->io, inbuf, &req.length );
-            node = tr_list_find( msgs->peerAskedFor, &req, peer_request_compare );
+            node = tr_list_find( msgs->peerAskedFor, &req, compareRequest );
             if( node != NULL ) {
                 void * data = node->data;
@@ -939 +1060 @@
         case BT_ALLOWED_FAST: {
             assert( msglen == 4 );
-            dbgmsg( msgs, "peer sent us a BT_ALLOWED_FAST" );
+            dbgmsg( msgs, "Got a BT_ALLOWED_FAST" );
             tr_peerIoReadUint32( msgs->io, inbuf, &ui32 );
             tr_bitfieldAdd( msgs->clientAllowedPieces, ui32 );
@@ -946 +1067 @@
             
         case BT_LTEP:
-            dbgmsg( msgs, "peer sent us a BT_LTEP" );
+            dbgmsg( msgs, "Got a BT_LTEP" );
             parseLtep( msgs, msglen, inbuf );
             break;
@@ -968 +1089 @@
     tor->downloadedCur += byteCount;
     msgs->info->pieceDataActivityDate = time( NULL );
-    tr_rcTransferred( msgs->info->rateToClient, byteCount );
+    tr_rcTransferred( msgs->info->rcToClient, byteCount );
     tr_rcTransferred( tor->download, byteCount );
     tr_rcTransferred( tor->handle->download, byteCount );
@@ -980 +1101 @@
     tor->uploadedCur += byteCount;
     msgs->info->pieceDataActivityDate = time( NULL );
-    tr_rcTransferred( msgs->info->rateToPeer, byteCount );
+    tr_rcTransferred( msgs->info->rcToPeer, byteCount );
     tr_rcTransferred( tor->upload, byteCount );
     tr_rcTransferred( tor->handle->upload, byteCount );
@@ -1055 +1176 @@
     key.length = length;
     req = (struct peer_request*) tr_list_remove( &msgs->clientAskedFor, &key,
-                                                 peer_request_compare );
+                                                 compareRequest );
     if( req == NULL ) {
         gotUnwantedBlock( msgs, index, offset, length );
@@ -1061 +1182 @@
         return;
     }
-    dbgmsg( msgs, "w00t peer sent us a block.  turnaround time was %d seconds", 
+    dbgmsg( msgs, "Got block %u:%u->%u (turnaround time %d secs)", 
+                     req->index, req->offset, req->length,
                      (int)(time(NULL) - req->time_requested) );
     tr_free( req );
@@ -1097 +1219 @@
 
     fireGotBlock( msgs, index, offset, length );
-    fireNeedReq( msgs );
 
     /**
@@ -1142 +1263 @@
     if( !msgs->blockToUs.length )
     {
-        dbgmsg( msgs, "w00t -- got block index %u, offset %u", msgs->blockToUs.index, msgs->blockToUs.offset );
+        dbgmsg( msgs, "got block %u:%u", msgs->blockToUs.index, msgs->blockToUs.offset );
         assert( (int)EVBUFFER_LENGTH( msgs->inBlock ) == tr_torBlockCountBytes( msgs->torrent, _tr_block(msgs->torrent,msgs->blockToUs.index, msgs->blockToUs.offset) ) );
         gotBlock( msgs, msgs->inBlock,
@@ -1220 +1341 @@
 
 static int
+ratePulse( void * vmsgs )
+{
+    tr_peermsgs * msgs = (tr_peermsgs *) vmsgs;
+    msgs->info->rateToClient = tr_rcRate( msgs->info->rcToClient );
+    msgs->info->rateToPeer = tr_rcRate( msgs->info->rcToPeer );
+    msgs->maxActiveRequests = MIN( 8 + (int)(msgs->info->rateToClient/10), 100 );
+    msgs->minActiveRequests = msgs->maxActiveRequests / 2;
+    return TRUE;
+}
+
+static int
 pulse( void * vmsgs )
 {
@@ -1233 +1365 @@
         tr_peerIoSetIOMode ( msgs->io, EV_READ, 0 );
     }
+
+    pumpRequestQueue( msgs );
 
     if( !canWrite( msgs ) )
@@ -1262 +1396 @@
             peerGotBytes( msgs, r->length );
 
-            dbgmsg( msgs, "putting req into out queue: index %d, offset %d, length %d ... %d blocks left in our queue", (int)r->index, (int)r->offset, (int)r->length, tr_list_size(msgs->peerAskedFor) );
+            dbgmsg( msgs, "Sending block %u:%u->%u (%d blocks left to send)", r->index, r->offset, r->length, tr_list_size(msgs->peerAskedFor) );
 
             tr_free( r );
@@ -1408 +1542 @@
         flags = tr_bencDictAdd( &val, "added.f" );
         tmp = walk = tr_new( uint8_t, diffs.addedCount );
-        for( i=0; i<diffs.addedCount; ++i ) {
-            dbgmsg( msgs, "PEX -->> -->> flag is %d", (int)diffs.added[i].flags );
+        for( i=0; i<diffs.addedCount; ++i )
             *walk++ = diffs.added[i].flags;
-        }
         assert( ( walk - tmp ) == diffs.addedCount );
         tr_bencInitStr( flags, tmp, walk-tmp, FALSE );
@@ -1478 +1610 @@
     m->info->have = tr_bitfieldNew( torrent->info.pieceCount );
     m->pulseTimer = tr_timerNew( m->handle, pulse, m, PEER_PULSE_INTERVAL );
+    m->rateTimer = tr_timerNew( m->handle, ratePulse, m, RATE_PULSE_INTERVAL );
     m->pexTimer = tr_timerNew( m->handle, pexPulse, m, PEX_INTERVAL );
     m->outMessages = evbuffer_new( );
@@ -1502 +1635 @@
     tr_peerIoSetIOFuncs( m->io, canRead, didWrite, gotError, m );
     tr_peerIoSetIOMode( m->io, EV_READ|EV_WRITE, 0 );
+    ratePulse( m );
 
     /**
@@ -1548 +1682 @@
     {
         tr_timerFree( &msgs->pulseTimer );
+        tr_timerFree( &msgs->rateTimer );
         tr_timerFree( &msgs->pexTimer );
         tr_publisherFree( &msgs->publisher );
+        tr_list_free( &msgs->clientWillAskFor, tr_free );
         tr_list_free( &msgs->clientAskedFor, tr_free );
         tr_list_free( &msgs->peerAskedFor, tr_free );

trunk/libtransmission/peer-msgs.h

@@ -66 +66 @@
     TR_PEERMSG_PEER_PROGRESS,
     TR_PEERMSG_GOT_ERROR,
+    TR_PEERMSG_CANCEL,
     TR_PEERMSG_NEED_REQ
 }

trunk/libtransmission/tracker.c

@@ -52 +52 @@
 
 /* the value of the 'numwant' argument passed in tracker requests */
-#define NUMWANT 75
+#define NUMWANT 128
 
 /* the length of the 'key' argument passed in tracker requests */

trunk/libtransmission/transmission.c

@@ -317 +317 @@
     tr_runInEventThread( h, tr_closeImpl, h );
     while( !h->isClosed )
-        tr_wait( 200 );
+        tr_wait( 100 );
 
     tr_eventClose( h );
-    while( h->events != NULL ) {
-        fprintf( stderr, "waiting for libevent thread to close...\n" );
-        tr_wait( 200 );
-    }
+    while( h->events != NULL )
+        tr_wait( 100 );
 
     tr_lockFree( h->lock );
     free( h->tag );
     free( h );
-    fprintf( stderr, "tr_close() completed.\n" );
+    fprintf( stderr, "libtransmission closed cleanly.\n" );
 }
 

trunk/libtransmission/utils.c

@@ -85 +85 @@
 }
 
+char*
+tr_getLogTimeStr( char * buf, int buflen )
+{
+    char tmp[64];
+    time_t now;
+    struct tm now_tm;
+    struct timeval tv;
+    int milliseconds;
+
+    now = time( NULL );
+    gettimeofday( &tv, NULL );
+
+    localtime_r( &now, &now_tm );
+    strftime( tmp, sizeof(tmp), "%H:%M:%S", &now_tm );
+    milliseconds = (int)(tv.tv_usec / 1000);
+    snprintf( buf, buflen, "%s.%03d", tmp, milliseconds );
+
+    return buf;
+}
+
 void
 tr_setMessageLevel( int level )

trunk/libtransmission/utils.h

@@ -38 +38 @@
 void tr_msg  ( int level, char * msg, ... );
 FILE* tr_getLog( void );
+
+char* tr_getLogTimeStr( char * buf, int buflen );
 
 int  tr_rand ( int );