Changeset 9494 for trunk/libtransmission/peer-mgr.c

Timestamp:

Nov 8, 2009, 11:20:00 PM (13 years ago)

Author:

charles

Message:

(trunk libT) #2548: T's request queue can send out too many duplicate requests

File:

• trunk/libtransmission/peer-mgr.c (modified) (19 diffs)

trunk/libtransmission/peer-mgr.c

@@ -137 +137 @@
 }
 
-struct tr_blockIterator
-{
-    time_t expirationDate;
-    struct tr_torrent_peers * t;
-    tr_block_index_t blockIndex, blockCount, *blocks;
-    tr_piece_index_t pieceIndex, pieceCount, *pieces;
+struct block_request
+{
+    tr_block_index_t block;
+    tr_peer * peer;
+    time_t sentAt;
+};
+
+struct weighted_piece
+{
+    tr_piece_index_t index;
+    int16_t salt;
+    int16_t requestCount;
 };
 
@@ -156 +162 @@
     tr_torrent               * tor;
     tr_peer                  * optimistic; /* the optimistic peer, or NULL if none */
-    struct tr_blockIterator  * refillQueue; /* used in refillPulse() */
     struct tr_peerMgr        * manager;
-    int                      * pendingRequestCount;
+    //int                      * pendingRequestCount;
 
     tr_bool                    isRunning;
+
+    struct block_request     * requests;
+    int                        requestsSort;
+    int                        requestCount;
+    int                        requestAlloc;
+
+    struct weighted_piece    * pieces;
+    int                        piecesSort;
+    int                        pieceCount;
+
+    tr_bool                    isInEndgame;
 }
 Torrent;
@@ -352 +368 @@
 }
 
-static void
-peerDestructor( tr_peer * peer )
-{
-    assert( peer );
+static void peerDeclinedAllRequests( Torrent *, const tr_peer * );
+
+static void
+peerDestructor( Torrent * t, tr_peer * peer )
+{
+    assert( peer != NULL );
+
+    peerDeclinedAllRequests( t, peer );
 
     if( peer->msgs != NULL )
@@ -386 +406 @@
     removed = tr_ptrArrayRemoveSorted( &t->peers, peer, peerCompare );
     assert( removed == peer );
-    peerDestructor( removed );
+    peerDestructor( t, removed );
 }
 
@@ -395 +415 @@
         removePeer( t, tr_ptrArrayNth( &t->peers, 0 ) );
 }
-
-static void blockIteratorFree( struct tr_blockIterator ** inout );
 
 static void
@@ -411 +429 @@
     evtimer_del( &t->refillTimer );
 
-    blockIteratorFree( &t->refillQueue );
     tr_ptrArrayDestruct( &t->webseeds, (PtrArrayForeachFunc)tr_webseedFree );
     tr_ptrArrayDestruct( &t->pool, (PtrArrayForeachFunc)tr_free );
@@ -417 +434 @@
     tr_ptrArrayDestruct( &t->peers, NULL );
 
-    tr_free( t->pendingRequestCount );
+    tr_free( t->requests );
+    tr_free( t->pieces );
     tr_free( t );
 }
 
 
-static void refillPulse( int, short, void* );
+//static void refillPulse( int, short, void* );
 
 static void peerCallbackFunc( void * vpeer,
@@ -442 +460 @@
     t->webseeds = TR_PTR_ARRAY_INIT;
     t->outgoingHandshakes = TR_PTR_ARRAY_INIT;
-    evtimer_set( &t->refillTimer, refillPulse, t );
+    t->requests = 0;
+    //evtimer_set( &t->refillTimer, refillPulse, t );
 
 
@@ -534 +553 @@
 }
 
-/****
-*****
-*****  REFILL
-*****
-****/
-
-static void
-assertValidPiece( Torrent * t, tr_piece_index_t piece )
-{
-    assert( t );
-    assert( t->tor );
-    assert( piece < t->tor->info.pieceCount );
-}
+/**
+***  REQUESTS
+***
+*** There are two data structures associated with managing block requests:
+*** 
+*** 1. Torrent::requests, an array of "struct block_request" which keeps
+***    track of which blocks have been requested, and when, and by which peers.
+***    This is list is used for (a) cancelling requests that have been pending
+***    for too long and (b) avoiding duplicate requests before endgame.
+*** 
+*** 2. Torrent::pieces, an array of "struct weighted_piece" which lists the
+***    pieces that we want to request.  It's used to decide which pieces to
+***    return next when tr_peerMgrGetBlockRequests() is called.
+**/ 
+
+/**
+*** struct block_request
+**/
+
+enum
+{
+    REQ_UNSORTED,
+    REQ_SORTED_BY_BLOCK,
+    REQ_SORTED_BY_TIME
+};
 
 static int
-getPieceRequests( Torrent * t, tr_piece_index_t piece )
-{
-    assertValidPiece( t, piece );
-
-    return t->pendingRequestCount ? t->pendingRequestCount[piece] : 0;
-}
-
-static void
-incrementPieceRequests( Torrent * t, tr_piece_index_t piece )
-{
-    assertValidPiece( t, piece );
-
-    if( t->pendingRequestCount == NULL )
-        t->pendingRequestCount = tr_new0( int, t->tor->info.pieceCount );
-    t->pendingRequestCount[piece]++;
-}
-
-static void
-decrementPieceRequests( Torrent * t, tr_piece_index_t piece )
-{
-    assertValidPiece( t, piece );
-
-    if( t->pendingRequestCount )
-        t->pendingRequestCount[piece]--;
-}
-
-struct tr_refill_piece
-{
-    tr_priority_t    priority;
-    uint32_t         piece;
-    uint32_t         peerCount;
-    int              random;
-    int              pendingRequestCount;
-    int              missingBlockCount;
+compareReqByBlock( const void * va, const void * vb )
+{
+    const struct block_request * a = va;
+    const struct block_request * b = vb;
+    if( a->block < b->block ) return -1;
+    if( a->block > b->block ) return 1;
+    return 0;
+}
+
+static int
+compareReqByTime( const void * va, const void * vb )
+{
+    const struct block_request * a = va;
+    const struct block_request * b = vb;
+    if( a->sentAt < b->sentAt ) return -1;
+    if( a->sentAt > b->sentAt ) return 1;
+    return 0;
+}
+
+static void
+requestListSort( Torrent * t, int mode )
+{
+    assert( mode==REQ_SORTED_BY_BLOCK || mode==REQ_SORTED_BY_TIME );
+
+    if( t->requestsSort != mode )
+    {
+        int(*compar)(const void *, const void *);
+
+        t->requestsSort = mode;
+
+        switch( mode ) {
+            case REQ_SORTED_BY_BLOCK: compar = compareReqByBlock; break;
+            case REQ_SORTED_BY_TIME: compar = compareReqByTime; break;
+            default: assert( 0 && "unhandled" );
+        }
+
+//fprintf( stderr, "sorting requests by %s\n", (mode==REQ_SORTED_BY_BLOCK)?"block":"time" ); 
+        qsort( t->requests, t->requestCount,
+               sizeof( struct block_request ), compar );
+    }
+}
+
+static void
+requestListAdd( Torrent * t, tr_block_index_t block, tr_peer * peer )
+{
+    struct block_request key;
+
+    /* ensure enough room is available... */
+    if( t->requestCount + 1 >= t->requestAlloc )
+    {
+        const int CHUNK_SIZE = 128;
+        t->requestAlloc += CHUNK_SIZE;
+        t->requests = tr_renew( struct block_request,
+                                t->requests, t->requestAlloc );
+    }
+
+    /* populate the record we're inserting */
+    key.block = block;
+    key.peer = peer;
+    key.sentAt = time( NULL );
+
+    /* insert the request to our array... */
+    switch( t->requestsSort )
+    {
+        case REQ_UNSORTED:
+        case REQ_SORTED_BY_TIME:
+            t->requests[t->requestCount++] = key;
+            break;
+
+        case REQ_SORTED_BY_BLOCK: {
+            tr_bool exact;
+            const int pos = tr_lowerBound( &key, t->requests, t->requestCount,
+                                           sizeof( struct block_request ),
+                                           compareReqByBlock, &exact );
+            assert( !exact );
+            memmove( t->requests + pos + 1,
+                     t->requests + pos,
+                     sizeof( struct block_request ) * ( t->requestCount++ - pos ) );
+            t->requests[pos] = key;
+            break;
+        }
+    }
+}
+
+static struct block_request *
+requestListLookup( Torrent * t, tr_block_index_t block )
+{
+    struct block_request key;
+    key.block = block;
+
+    requestListSort( t, REQ_SORTED_BY_BLOCK );
+
+    return bsearch( &key, t->requests, t->requestCount,
+                    sizeof( struct block_request ),
+                    compareReqByBlock );
+}
+
+static void
+requestListRemove( Torrent * t, tr_block_index_t block )
+{
+    const struct block_request * b = requestListLookup( t, block );
+    if( b != NULL )
+    {
+        const int pos = b - t->requests;
+        assert( pos < t->requestCount );
+        memmove( t->requests + pos,
+                 t->requests + pos + 1,
+                 sizeof( struct block_request ) * ( --t->requestCount - pos ) );
+    }
+}
+
+/**
+*** struct weighted_piece
+**/
+
+enum
+{
+    PIECES_UNSORTED,
+    PIECES_SORTED_BY_INDEX,
+    PIECES_SORTED_BY_WEIGHT
 };
 
+const tr_torrent * weightTorrent;
+
+/* we try to create a "weight" s.t. high-priority pieces come before others,
+ * and that partially-complete pieces come before empty ones. */
 static int
-compareRefillPiece( const void * aIn, const void * bIn )
-{
-    const struct tr_refill_piece * a = aIn;
-    const struct tr_refill_piece * b = bIn;
-
-    /* if one piece has a higher priority, it goes first */
-    if( a->priority != b->priority )
-        return a->priority > b->priority ? -1 : 1;
-
-    /* have a per-priority endgame */
-    if( a->pendingRequestCount != b->pendingRequestCount )
-        return a->pendingRequestCount < b->pendingRequestCount ? -1 : 1;
-
-    /* fewer missing pieces goes first */
-    if( a->missingBlockCount != b->missingBlockCount )
-        return a->missingBlockCount < b->missingBlockCount ? -1 : 1;
-
-    /* otherwise if one has fewer peers, it goes first */
-    if( a->peerCount != b->peerCount )
-        return a->peerCount < b->peerCount ? -1 : 1;
-
-    /* otherwise go with our random seed */
-    if( a->random != b->random )
-        return a->random < b->random ? -1 : 1;
-
+comparePieceByWeight( const void * va, const void * vb )
+{
+    const struct weighted_piece * a = va;
+    const struct weighted_piece * b = vb;
+    int ia, ib, missing, pending;
+    const tr_torrent * tor = weightTorrent;
+
+    /* primary key: weight */
+    missing = tr_cpMissingBlocksInPiece( &tor->completion, a->index );
+    pending = a->requestCount;
+    ia = missing > pending ? missing - pending : (int)(tor->blockCountInPiece + pending);
+    missing = tr_cpMissingBlocksInPiece( &tor->completion, b->index );
+    pending = b->requestCount;
+    ib = missing > pending ? missing - pending : (int)(tor->blockCountInPiece + pending);
+    if( ia < ib ) return -1;
+    if( ia > ib ) return 1;
+
+    /* secondary key: higher priorities go first */
+    ia = tor->info.pieces[a->index].priority;
+    ib = tor->info.pieces[b->index].priority;
+    if( ia > ib ) return -1;
+    if( ia < ib ) return 1;
+
+    /* tertiary key: random */
+    return a->salt - b->salt;
+}
+
+static int
+comparePieceByIndex( const void * va, const void * vb )
+{
+    const struct weighted_piece * a = va;
+    const struct weighted_piece * b = vb;
+    if( a->index < b->index ) return -1;
+    if( a->index > b->index ) return 1;
     return 0;
 }
 
-static tr_piece_index_t *
-getPreferredPieces( Torrent * t, tr_piece_index_t * pieceCount )
-{
-    const tr_torrent  * tor = t->tor;
-    const tr_info     * inf = &tor->info;
-    tr_piece_index_t    i;
-    tr_piece_index_t    poolSize = 0;
-    tr_piece_index_t  * pool = tr_new( tr_piece_index_t , inf->pieceCount );
-    int                 peerCount;
-    const tr_peer    ** peers;
-
-    assert( torrentIsLocked( t ) );
-
-    peers = (const tr_peer**) tr_ptrArrayBase( &t->peers );
-    peerCount = tr_ptrArraySize( &t->peers );
-
-    /* make a list of the pieces that we want but don't have */
-    for( i = 0; i < inf->pieceCount; ++i )
-        if( !tor->info.pieces[i].dnd
-                && !tr_cpPieceIsComplete( &tor->completion, i ) )
-            pool[poolSize++] = i;
-
-    /* sort the pool by which to request next */
-    if( poolSize > 1 )
-    {
-        tr_piece_index_t j;
-        struct tr_refill_piece * p = tr_new( struct tr_refill_piece, poolSize );
-
-        for( j = 0; j < poolSize; ++j )
+static void
+pieceListSort( Torrent * t, int mode )
+{
+    int(*compar)(const void *, const void *);
+
+    assert( mode==PIECES_SORTED_BY_INDEX
+         || mode==PIECES_SORTED_BY_WEIGHT );
+
+    if( t->piecesSort != mode )
+    {
+//fprintf( stderr, "sort mode was %d, is now %d\n", t->piecesSort, mode );
+        t->piecesSort = mode;
+
+        switch( mode ) {
+            case PIECES_SORTED_BY_WEIGHT: compar = comparePieceByWeight; break;
+            case PIECES_SORTED_BY_INDEX: compar = comparePieceByIndex; break;
+            default: assert( 0 && "unhandled" );  break;
+        }
+
+//fprintf( stderr, "sorting pieces by %s...\n", (mode==PIECES_SORTED_BY_WEIGHT)?"weight":"index" );
+        weightTorrent = t->tor;
+        qsort( t->pieces, t->pieceCount,
+               sizeof( struct weighted_piece ), compar );
+    }
+
+    /* Also, as long as we've got the pieces sorted by weight,
+     * let's also update t.isInEndgame */
+    if( t->piecesSort == PIECES_SORTED_BY_WEIGHT )
+    {
+        tr_bool endgame = TRUE;
+
+        if( ( t->pieces != NULL ) && ( t->pieceCount > 0 ) )
         {
-            int k;
-            const tr_piece_index_t piece = pool[j];
-            struct tr_refill_piece * setme = p + j;
-
-            setme->piece = piece;
-            setme->priority = inf->pieces[piece].priority;
-            setme->peerCount = 0;
-            setme->random = tr_cryptoWeakRandInt( INT_MAX );
-            setme->pendingRequestCount = getPieceRequests( t, piece );
-            setme->missingBlockCount
-                         = tr_cpMissingBlocksInPiece( &tor->completion, piece );
-
-            for( k = 0; k < peerCount; ++k )
+            const tr_completion * cp = &t->tor->completion;
+            const struct weighted_piece * p = t->pieces;
+            const int pending = p->requestCount;
+            const int missing = tr_cpMissingBlocksInPiece( cp, p->index );
+            endgame = pending >= missing;
+        }
+
+        t->isInEndgame = endgame;
+    }
+}
+
+static struct weighted_piece *
+pieceListLookup( Torrent * t, tr_piece_index_t index )
+{
+    struct weighted_piece key;
+    key.index = index;
+
+    pieceListSort( t, PIECES_SORTED_BY_INDEX );
+
+    return bsearch( &key, t->pieces, t->pieceCount,
+                    sizeof( struct weighted_piece ),
+                    comparePieceByIndex );
+}
+
+static void
+pieceListRebuild( Torrent * t )
+{
+    if( !tr_torrentIsSeed( t->tor ) )
+    {
+        tr_piece_index_t i;
+        tr_piece_index_t * pool;
+        tr_piece_index_t poolCount = 0;
+        const tr_torrent * tor = t->tor;
+        const tr_info * inf = tr_torrentInfo( tor );
+        struct weighted_piece * pieces;
+        int pieceCount;
+
+        /* build the new list */
+        pool = tr_new( tr_piece_index_t, inf->pieceCount );
+        for( i=0; i<inf->pieceCount; ++i )
+            if( !inf->pieces[i].dnd )
+                if( !tr_cpPieceIsComplete( &tor->completion, i ) )
+                    pool[poolCount++] = i;
+        pieceCount = poolCount;
+        pieces = tr_new0( struct weighted_piece, pieceCount );
+        for( i=0; i<poolCount; ++i ) {
+            struct weighted_piece * piece = pieces + i;
+            piece->index = pool[i];
+            piece->requestCount = 0;
+            piece->salt = tr_cryptoWeakRandInt( 255 );
+        }
+
+        /* if we already had a list of pieces, merge it into
+         * the new list so we don't lose its requestCounts */
+        if( t->pieces != NULL )
+        {
+            struct weighted_piece * o = t->pieces;
+            struct weighted_piece * oend = o + t->pieceCount;
+            struct weighted_piece * n = pieces;
+            struct weighted_piece * nend = n + pieceCount;
+
+            pieceListSort( t, PIECES_SORTED_BY_INDEX );
+
+            while( o!=oend && n!=nend ) {
+                if( o->index < n->index )
+                    ++o;
+                else if( o->index > n->index )
+                    ++n;
+                else
+                    *n++ = *o++;
+            }
+
+            tr_free( t->pieces );
+        }
+
+        t->pieces = pieces;
+        t->pieceCount = pieceCount;
+        t->piecesSort = PIECES_SORTED_BY_INDEX;
+
+        /* cleanup */
+        tr_free( pool );
+    }
+}
+
+static void
+pieceListRemovePiece( Torrent * t, tr_piece_index_t piece )
+{
+    struct weighted_piece * p = pieceListLookup( t, piece );
+
+    if( p != NULL )
+    {
+        const int pos = p - t->pieces;
+
+        memmove( t->pieces + pos,
+                 t->pieces + pos + 1,
+                 sizeof( struct weighted_piece ) * ( --t->pieceCount - pos ) );
+
+        if( t->pieceCount == 0 )
+        {
+            tr_free( t->pieces );
+            t->pieces = NULL;
+        }
+    }
+}
+
+static void
+pieceListRemoveRequest( Torrent * t, tr_block_index_t block )
+{
+    struct weighted_piece * p;
+    const tr_piece_index_t index = tr_torBlockPiece( t->tor, block );
+
+    if(( p = pieceListLookup( t, index )))
+        if( p->requestCount > 0 )
+            --p->requestCount;
+
+    /* note: this invalidates the weighted.piece.weight field,
+     * but that's OK since the call to pieceListLookup ensured
+     * that we were sorted by index anyway.. next time we resort
+     * by weight, pieceListSort() will update the weights */
+}
+
+/**
+***
+**/
+
+void
+tr_peerMgrRebuildRequests( tr_torrent * tor )
+{
+    assert( tr_isTorrent( tor ) );
+
+    pieceListRebuild( tor->torrentPeers );
+}
+
+void
+tr_peerMgrGetNextRequests( tr_torrent           * tor,
+                           tr_peer              * peer,
+                           int                    numwant,
+                           tr_block_index_t     * setme,
+                           int                  * numgot )
+{
+    int i;
+    int got;
+    Torrent * t;
+    struct weighted_piece * pieces;
+    const tr_bitfield * have = peer->have;
+
+    /* sanity clause */
+    assert( tr_isTorrent( tor ) );
+    assert( numwant > 0 );
+
+    /* walk through the pieces and find blocks that should be requested */
+    got = 0;
+    t = tor->torrentPeers;
+
+    /* prep the pieces list */
+    if( t->pieces == NULL )
+        pieceListRebuild( t );
+    pieceListSort( t, PIECES_SORTED_BY_WEIGHT );
+//if( t->isInEndgame ) fprintf( stderr, "endgame\n" );
+
+#if 0
+{
+int i=0, n=MIN(10,t->pieceCount);
+fprintf( stderr, "the next pieces we want to request are " );
+for( i=0; i<n; i++ ) fprintf( stderr, "%d(weight:%d) ", (int)t->pieces[i].index, (int)t->pieces[i].weight );
+fprintf( stderr, "\n" );
+}
+#endif
+
+    pieces = t->pieces;
+    for( i=0; i<t->pieceCount && got<numwant; ++i )
+    {
+        struct weighted_piece * p = pieces + i;
+
+        /* if the peer has this piece that we want... */
+        if( tr_bitfieldHasFast( have, p->index ) )
+        {
+            tr_block_index_t b = tr_torPieceFirstBlock( tor, p->index );
+            const tr_block_index_t e = b + tr_torPieceCountBlocks( tor, p->index );
+
+            for( ; b!=e && got<numwant; ++b )
             {
-                const tr_peer * peer = peers[k];
-                if( peer->peerIsInterested
-                        && !peer->clientIsChoked
-                        && tr_bitfieldHas( peer->have, piece ) )
-                    ++setme->peerCount;
+                struct block_request * breq;
+
+                /* don't request blocks we've already got */
+                if( tr_cpBlockIsCompleteFast( &tor->completion, b ) )
+                    continue;
+
+                /* don't request blocks we've already requested (FIXME) */
+                breq = requestListLookup( t, b );
+                if( breq != NULL ) {
+                    assert( breq->peer != NULL );
+                    if( breq->peer == peer ) continue;
+                    if( !t->isInEndgame ) continue;
+                }
+
+                setme[got++] = b;
+//fprintf( stderr, "peer %p is requesting block %"PRIu64"\n", peer, b );
+
+                /* update our own tables */
+                if( breq == NULL )
+                    requestListAdd( t, b, peer );
+                ++p->requestCount;
             }
         }
-
-        qsort( p, poolSize, sizeof( struct tr_refill_piece ),
-               compareRefillPiece );
-
-        for( j = 0; j < poolSize; ++j )
-            pool[j] = p[j].piece;
-
-        tr_free( p );
-    }
-
-    *pieceCount = poolSize;
-    return pool;
-}
-
-static struct tr_blockIterator*
-blockIteratorNew( Torrent * t )
-{
-    struct tr_blockIterator * i = tr_new0( struct tr_blockIterator, 1 );
-    i->expirationDate = time( NULL ) + PIECE_LIST_SHELF_LIFE_SECS;
-    i->t = t;
-    i->pieces = getPreferredPieces( t, &i->pieceCount );
-    i->blocks = tr_new0( tr_block_index_t, t->tor->blockCountInPiece );
-    tordbg( t, "creating new refill queue.. it contains %"PRIu32" pieces", i->pieceCount );
-    return i;
-}
-
-static tr_bool
-blockIteratorNext( struct tr_blockIterator * i, tr_block_index_t * setme )
-{
-    tr_bool found;
-    Torrent * t = i->t;
-    tr_torrent * tor = t->tor;
-
-    while( ( i->blockIndex == i->blockCount )
-        && ( i->pieceIndex < i->pieceCount ) )
-    {
-        const tr_piece_index_t index = i->pieces[i->pieceIndex++];
-        const tr_block_index_t b = tr_torPieceFirstBlock( tor, index );
-        const tr_block_index_t e = b + tr_torPieceCountBlocks( tor, index );
-        tr_block_index_t block;
-
-        assert( index < tor->info.pieceCount );
-
-        i->blockCount = 0;
-        i->blockIndex = 0;
-        for( block=b; block!=e; ++block )
-            if( !tr_cpBlockIsCompleteFast( &tor->completion, block ) )
-                i->blocks[i->blockCount++] = block;
-    }
-
-    assert( i->blockCount <= tor->blockCountInPiece );
-
-    if(( found = ( i->blockIndex < i->blockCount )))
-        *setme = i->blocks[i->blockIndex++];
-
-    return found;
-}
-
-static void
-blockIteratorSkipCurrentPiece( struct tr_blockIterator * i )
-{
-    i->blockIndex = i->blockCount;
-}
-
-static void
-blockIteratorFree( struct tr_blockIterator ** inout )
-{
-    struct tr_blockIterator * it = *inout;
-
-    if( it != NULL )
-    {
-        tr_free( it->blocks );
-        tr_free( it->pieces );
-        tr_free( it );
-    }
-
-    *inout = NULL;
-}
-
-static tr_peer**
-getPeersUploadingToClient( Torrent * t,
-                           int *     setmeCount )
-{
-    int j;
-    int peerCount = 0;
-    int retCount = 0;
-    tr_peer ** peers = (tr_peer **) tr_ptrArrayPeek( &t->peers, &peerCount );
-    tr_peer ** ret = tr_new( tr_peer *, peerCount );
-
-    j = 0; /* this is a temporary test to make sure we walk through all the peers */
-    if( peerCount )
-    {
-        /* Get a list of peers we're downloading from.
-           Pick a different starting point each time so all peers
-           get a chance at being the first in line */
-        const int fencepost = tr_cryptoWeakRandInt( peerCount );
-        int i = fencepost;
-        do {
-            if( clientIsDownloadingFrom( peers[i] ) )
-                ret[retCount++] = peers[i];
-            i = ( i + 1 ) % peerCount;
-            ++j;
-        } while( i != fencepost );
-    }
-    assert( j == peerCount );
-    *setmeCount = retCount;
-    return ret;
-}
-
-static uint32_t
-getBlockOffsetInPiece( const tr_torrent * tor, uint64_t b )
-{
-    const uint64_t piecePos = tor->info.pieceSize * tr_torBlockPiece( tor, b );
-    const uint64_t blockPos = tor->blockSize * b;
-    assert( blockPos >= piecePos );
-    return (uint32_t)( blockPos - piecePos );
-}
-
+    }
+
+    /* We almost always change only a handful of pieces in the array.
+     * In these cases, it's cheaper to sort those changed pieces and merge,
+     * than qsort()ing the whole array again */
+    if( got > 0 )
+    {
+        struct weighted_piece * p;
+        struct weighted_piece * pieces;
+        struct weighted_piece * a = t->pieces;
+        struct weighted_piece * a_end = t->pieces + i;
+        struct weighted_piece * b = a_end;
+        struct weighted_piece * b_end = t->pieces + t->pieceCount;
+
+        /* rescore the pieces that we changed */
+        weightTorrent = t->tor;
+//fprintf( stderr, "sorting %d changed pieces...\n", (int)(a_end-a) );
+        qsort( a, a_end-a, sizeof( struct weighted_piece ), comparePieceByWeight );
+
+        /* allocate a new array */
+        p = pieces = tr_new( struct weighted_piece, t->pieceCount );
+
+        /* merge the two sorted arrays into this new array */
+        weightTorrent = t->tor;
+        while( a!=a_end && b!=b_end )
+            *p++ = comparePieceByWeight( a, b ) < 0 ? *a++ : *b++;
+        while( a!=a_end ) *p++ = *a++;
+        while( b!=b_end ) *p++ = *b++;
+
+#if 0
+        /* make sure we did it right */
+        assert( p - pieces == t->pieceCount );
+        for( it=pieces; it+1<p; ++it )
+            assert( it->weight <= it[1].weight );
+#endif
+
+        /* update */
+        tr_free( t->pieces );
+        t->pieces = pieces;
+    }
+
+    //fprintf( stderr, "peer %p wanted %d requests; got %d\n", peer, numwant, got );
+    *numgot = got;
+}
+
+tr_bool
+tr_peerMgrDidPeerRequest( const tr_torrent  * tor,
+                          const tr_peer     * peer,
+                          tr_block_index_t    block )
+{
+    const Torrent * t = tor->torrentPeers;
+    const struct block_request * b = requestListLookup( (Torrent*)t, block );
+    if( b == NULL ) return FALSE;
+    if( b->peer == peer ) return TRUE;
+    if( t->isInEndgame ) return TRUE;
+    return FALSE;
+}
+
+/* cancel requests that are too old */
 static int
 refillUpkeep( void * vmgr )
 {
-    tr_torrent * tor = NULL;
+    time_t now;
+    time_t too_old;
+    tr_torrent * tor;
     tr_peerMgr * mgr = vmgr;
-    time_t now;
     managerLock( mgr );
 
     now = time( NULL );
-    while(( tor = tr_torrentNext( mgr->session, tor ))) {
+    too_old = now - REQUEST_TTL_SECS;
+
+    tor = NULL;
+    while(( tor = tr_torrentNext( mgr->session, tor )))
+    {
         Torrent * t = tor->torrentPeers;
-        if( t && t->refillQueue && ( t->refillQueue->expirationDate <= now ) ) {
-            tordbg( t, "refill queue is past its shelf date; discarding." );
-            blockIteratorFree( &t->refillQueue );
+        const int n = t->requestCount;
+        if( n > 0 )
+        {
+            int keepCount = 0;
+            int cancelCount = 0;
+            struct block_request * keep = tr_new( struct block_request, n );
+            struct block_request * cancel = tr_new( struct block_request, n );
+            const struct block_request * it;
+            const struct block_request * end;
+
+            for( it=t->requests, end=it+n; it!=end; ++it )
+                if( it->sentAt <= too_old )
+                    cancel[cancelCount++] = *it;
+                else
+                    keep[keepCount++] = *it;
+
+            /* prune out the ones we aren't keeping */
+            tr_free( t->requests );
+            t->requests = keep;
+            t->requestCount = keepCount;
+            t->requestAlloc = n;
+
+            /* send cancel messages for all the "cancel" ones */
+            for( it=cancel, end=it+cancelCount; it!=end; ++it )
+                if( ( it->peer != NULL ) && ( it->peer->msgs != NULL ) )
+                    tr_peerMsgsCancel( it->peer->msgs, it->block );
+
+            /* decrement the pending request counts for the timed-out blocks */
+            for( it=cancel, end=it+cancelCount; it!=end; ++it )
+                pieceListRemoveRequest( t, it->block );
+
+            /* cleanup loop */
+            tr_free( cancel );
         }
     }
@@ -802 +1095 @@
     managerUnlock( mgr );
     return TRUE;
-}
-
-static void
-sortPeersByLiveliness( tr_peer ** peers, void ** clientData, int n, uint64_t now );
-
-static void
-refillPulse( int fd UNUSED, short type UNUSED, void * vtorrent )
-{
-    tr_block_index_t block;
-    int peerCount;
-    int webseedCount;
-    tr_peer ** peers;
-    tr_webseed ** webseeds;
-    Torrent * t = vtorrent;
-    tr_torrent * tor = t->tor;
-    tr_bool hasNext = TRUE;
-
-    if( !t->isRunning )
-        return;
-    if( tr_torrentIsSeed( t->tor ) )
-        return;
-
-    torrentLock( t );
-    tordbg( t, "Refilling Request Buffers..." );
-
-    if( t->refillQueue == NULL )
-        t->refillQueue = blockIteratorNew( t );
-
-    peers = getPeersUploadingToClient( t, &peerCount );
-    sortPeersByLiveliness( peers, NULL, peerCount, tr_date( ) );
-    webseedCount = tr_ptrArraySize( &t->webseeds );
-    webseeds = tr_memdup( tr_ptrArrayBase( &t->webseeds ),
-                          webseedCount * sizeof( tr_webseed* ) );
-
-    while( ( webseedCount || peerCount )
-        && (( hasNext = blockIteratorNext( t->refillQueue, &block ))) )
-    {
-        int j;
-        tr_bool handled = FALSE;
-
-        const tr_piece_index_t index = tr_torBlockPiece( tor, block );
-        const uint32_t offset = getBlockOffsetInPiece( tor, block );
-        const uint32_t length = tr_torBlockCountBytes( tor, block );
-
-        assert( block < tor->blockCount );
-
-        /* find a peer who can ask for this block */
-        for( j=0; !handled && j<peerCount; )
-        {
-            const tr_addreq_t val = tr_peerMsgsAddRequest( peers[j]->msgs, index, offset, length );
-            switch( val )
-            {
-                case TR_ADDREQ_FULL:
-                case TR_ADDREQ_CLIENT_CHOKED:
-                    peers[j] = peers[--peerCount];
-                    break;
-
-                case TR_ADDREQ_MISSING:
-                case TR_ADDREQ_DUPLICATE:
-                    ++j;
-                    break;
-
-                case TR_ADDREQ_OK:
-                    incrementPieceRequests( t, index );
-                    handled = TRUE;
-                    break;
-
-                default:
-                    assert( 0 && "unhandled value" );
-                    break;
-            }
-        }
-
-        /* maybe one of the webseeds can do it */
-        for( j=0; !handled && j<webseedCount; )
-        {
-            const tr_addreq_t val = tr_webseedAddRequest( webseeds[j], index, offset, length );
-            switch( val )
-            {
-                case TR_ADDREQ_FULL:
-                    webseeds[j] = webseeds[--webseedCount];
-                    break;
-
-                case TR_ADDREQ_OK:
-                    incrementPieceRequests( t, index );
-                    handled = TRUE;
-                    break;
-
-                default:
-                    assert( 0 && "unhandled value" );
-                    break;
-            }
-        }
-
-        if( !handled )
-            blockIteratorSkipCurrentPiece( t->refillQueue );
-    }
-
-    /* cleanup */
-    tr_free( webseeds );
-    tr_free( peers );
-
-    if( !hasNext ) {
-        tordbg( t, "refill queue has no more blocks to request... freeing (webseed count: %d, peer count: %d)", webseedCount, peerCount );
-        blockIteratorFree( &t->refillQueue );
-    }
-
-    torrentUnlock( t );
-}
-
-static void
-broadcastGotBlock( Torrent * t, uint32_t index, uint32_t offset, uint32_t length )
-{
-    size_t i;
-    size_t peerCount;
-    tr_peer ** peers;
-
-    assert( torrentIsLocked( t ) );
-
-    tordbg( t, "got a block; cancelling any duplicate requests from peers %"PRIu32":%"PRIu32"->%"PRIu32, index, offset, length );
-
-    peerCount = tr_ptrArraySize( &t->peers );
-    peers = (tr_peer**) tr_ptrArrayBase( &t->peers );
-    for( i=0; i<peerCount; ++i )
-        if( peers[i]->msgs )
-            tr_peerMsgsCancel( peers[i]->msgs, index, offset, length );
 }
 
@@ -946 +1113 @@
 
 static void
-gotBadPiece( Torrent *        t,
-             tr_piece_index_t pieceIndex )
+gotBadPiece( Torrent * t, tr_piece_index_t pieceIndex )
 {
     tr_torrent *   tor = t->tor;
@@ -954 +1120 @@
     tor->corruptCur += byteCount;
     tor->downloadedCur -= MIN( tor->downloadedCur, byteCount );
-}
-
-static void
-refillSoon( Torrent * t )
-{
-    if( !evtimer_pending( &t->refillTimer, NULL ) )
-        tr_timerAdd( &t->refillTimer, 0, REFILL_PERIOD_MSEC );
 }
 
@@ -1009 +1168 @@
     }
 #endif
+}
+
+static void
+decrementDownloadedCount( tr_torrent * tor, uint32_t byteCount )
+{
+    tor->downloadedCur -= MIN( tor->downloadedCur, byteCount );
+}
+
+static void
+clientGotUnwantedBlock( tr_torrent * tor, tr_block_index_t block )
+{
+    decrementDownloadedCount( tor, tr_torBlockCountBytes( tor, block ) );
+}
+
+static void
+removeRequestFromTables( Torrent * t, tr_block_index_t block )
+{
+    requestListRemove( t, block );
+    pieceListRemoveRequest( t, block );
+}
+
+/* peer choked us, or maybe it disconnected.
+   either way we need to remove all its requests */
+static void
+peerDeclinedAllRequests( Torrent * t, const tr_peer * peer )
+{
+    int i, n;
+    tr_block_index_t * blocks = tr_new( tr_block_index_t, t->requestCount );
+
+    for( i=n=0; i<t->requestCount; ++i )
+        if( peer == t->requests[i].peer )
+            blocks[n++] = t->requests[i].block;
+
+    for( i=0; i<n; ++i )
+        removeRequestFromTables( t, blocks[i] );
+
+    tr_free( blocks );
 }
 
@@ -1035 +1231 @@
             break;
 
-        case TR_PEER_NEED_REQ:
-            refillSoon( t );
-            break;
-
-        case TR_PEER_CANCEL:
-            decrementPieceRequests( t, e->pieceIndex );
-            break;
-
         case TR_PEER_PEER_GOT_DATA:
         {
@@ -1067 +1255 @@
             break;
         }
+
+        case TR_PEER_CLIENT_GOT_REJ:
+            removeRequestFromTables( t, _tr_block( t->tor, e->pieceIndex, e->offset ) );
+            break;
+
+        case TR_PEER_CLIENT_GOT_CHOKE:
+            peerDeclinedAllRequests( t, peer );
+            break;
 
         case TR_PEER_CLIENT_GOT_PORT:
@@ -1129 +1325 @@
         {
             tr_torrent * tor = t->tor;
-
             tr_block_index_t block = _tr_block( tor, e->pieceIndex, e->offset );
-
-            tr_cpBlockAdd( &tor->completion, block );
-            tr_torrentSetDirty( tor );
-            decrementPieceRequests( t, e->pieceIndex );
-
-            broadcastGotBlock( t, e->pieceIndex, e->offset, e->length );
-
-            if( tr_cpPieceIsComplete( &tor->completion, e->pieceIndex ) )
+//static int numBlocks = 0;
+//fprintf( stderr, "got a total of %d blocks\n", ++numBlocks );
+
+            requestListRemove( t, block );
+            pieceListRemoveRequest( t, block );
+
+            if( tr_cpBlockIsComplete( &tor->completion, block ) )
             {
-                const tr_piece_index_t p = e->pieceIndex;
-                const tr_bool ok = tr_ioTestPiece( tor, p, NULL, 0 );
-
-                if( !ok )
+                tordbg( t, "we have this block already..." );
+                clientGotUnwantedBlock( tor, block );
+            }
+            else
+            {
+                tr_cpBlockAdd( &tor->completion, block );
+                tr_torrentSetDirty( tor );
+
+                if( tr_cpPieceIsComplete( &tor->completion, e->pieceIndex ) )
                 {
-                    tr_torerr( tor, _( "Piece %lu, which was just downloaded, failed its checksum test" ),
-                               (unsigned long)p );
-                }
-
-                tr_torrentSetHasPiece( tor, p, ok );
-                tr_torrentSetPieceChecked( tor, p, TRUE );
-                tr_peerMgrSetBlame( tor, p, ok );
-
-                if( !ok )
-                {
-                    gotBadPiece( t, p );
-                }
-                else
-                {
-                    int i;
-                    int peerCount;
-                    tr_peer ** peers;
-                    tr_file_index_t fileIndex;
-
-                    peerCount = tr_ptrArraySize( &t->peers );
-                    peers = (tr_peer**) tr_ptrArrayBase( &t->peers );
-                    for( i=0; i<peerCount; ++i )
-                        tr_peerMsgsHave( peers[i]->msgs, p );
-
-                    for( fileIndex=0; fileIndex<tor->info.fileCount; ++fileIndex ) {
-                        const tr_file * file = &tor->info.files[fileIndex];
-                        if( ( file->firstPiece <= p ) && ( p <= file->lastPiece ) )
-                            if( tr_cpFileIsComplete( &tor->completion, fileIndex ) )
-                                tr_torrentFileCompleted( tor, fileIndex );
+                    const tr_piece_index_t p = e->pieceIndex;
+                    const tr_bool ok = tr_ioTestPiece( tor, p, NULL, 0 );
+//fprintf( stderr, "we now have piece #%d\n", (int)p );
+
+                    if( !ok )
+                    {
+                        tr_torerr( tor, _( "Piece %lu, which was just downloaded, failed its checksum test" ),
+                                   (unsigned long)p );
+                    }
+
+                    tr_torrentSetHasPiece( tor, p, ok );
+                    tr_torrentSetPieceChecked( tor, p, TRUE );
+                    tr_peerMgrSetBlame( tor, p, ok );
+
+                    if( !ok )
+                    {
+                        gotBadPiece( t, p );
+                    }
+                    else
+                    {
+                        int i;
+                        int peerCount;
+                        tr_peer ** peers;
+                        tr_file_index_t fileIndex;
+
+                        peerCount = tr_ptrArraySize( &t->peers );
+                        peers = (tr_peer**) tr_ptrArrayBase( &t->peers );
+                        for( i=0; i<peerCount; ++i )
+                            tr_peerMsgsHave( peers[i]->msgs, p );
+
+                        for( fileIndex=0; fileIndex<tor->info.fileCount; ++fileIndex ) {
+                            const tr_file * file = &tor->info.files[fileIndex];
+                            if( ( file->firstPiece <= p ) && ( p <= file->lastPiece ) )
+                                if( tr_cpFileIsComplete( &tor->completion, fileIndex ) )
+                                    tr_torrentFileCompleted( tor, fileIndex );
+                        }
+
+                        pieceListRemovePiece( t, p );
                     }
                 }
@@ -1635 +1843 @@
     ensureMgrTimersExist( t->manager );
 
-    if( !t->isRunning )
-    {
-        t->isRunning = TRUE;
-
-        if( !tr_ptrArrayEmpty( &t->webseeds ) )
-            refillSoon( t );
-    }
+    t->isRunning = TRUE;
 
     rechokePulse( t->manager );
@@ -1650 +1852 @@
 stopTorrent( Torrent * t )
 {
+    int i, n;
+
     assert( torrentIsLocked( t ) );
 
@@ -1655 +1859 @@
 
     /* disconnect the peers. */
-    tr_ptrArrayForeach( &t->peers, (PtrArrayForeachFunc)peerDestructor );
+    for( i=0, n=tr_ptrArraySize( &t->peers ); i<n; ++i )
+        peerDestructor( t, tr_ptrArrayNth( &t->peers, i ) );
     tr_ptrArrayClear( &t->peers );