Changeset 9494

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

Location:

trunk/libtransmission

Files:

• Makefile.am (modified) (4 diffs)
• peer-common.h (modified) (2 diffs)
• peer-mgr.c (modified) (19 diffs)
• peer-mgr.h (modified) (2 diffs)
• peer-msgs.c (modified) (28 diffs)
• peer-msgs.h (modified) (1 diff)
• request-list-test.c (deleted)
• request-list.c (deleted)
• request-list.h (deleted)
• torrent.c (modified) (2 diffs)
• webseed.c (modified) (1 diff)

trunk/libtransmission/Makefile.am

@@ -40 +40 @@
     publish.c \
     ratecontrol.c \
-    request-list.c \
     resume.c \
     rpcimpl.c \
@@ -88 +87 @@
     publish.h \
     ratecontrol.h \
-    request-list.h \
     resume.h \
     rpcimpl.h \
@@ -112 +110 @@
     json-test \
     peer-msgs-test \
-    request-list-test \
     rpc-test \
     test-peer-id \
@@ -158 +155 @@
 test_peer_id_LDFLAGS = ${apps_ldflags}
 
-request_list_test_SOURCES = request-list-test.c
-request_list_test_LDADD = ${apps_ldadd}
-request_list_test_LDFLAGS = ${apps_ldflags}
-
 peer_msgs_test_SOURCES = peer-msgs-test.c
 peer_msgs_test_LDADD = ${apps_ldadd}

trunk/libtransmission/peer-common.h

@@ -29 +29 @@
 #include "transmission.h"
 
+enum
+{
+    /** when we're making requests from another peer, 
+        batch them together to send enough requests to
+        meet our bandwidth goals for the next N seconds */
+    REQUEST_BUF_SECS = 10,
+
+    /** how long we'll let requests we've made stay pending
+        before we cancel them */
+    REQUEST_TTL_SECS = 30
+};
+
 typedef enum
 {
@@ -46 +58 @@
 {
     TR_PEER_CLIENT_GOT_BLOCK,
+    TR_PEER_CLIENT_GOT_CHOKE,
     TR_PEER_CLIENT_GOT_DATA,
     TR_PEER_CLIENT_GOT_ALLOWED_FAST,
     TR_PEER_CLIENT_GOT_SUGGEST,
     TR_PEER_CLIENT_GOT_PORT,
+    TR_PEER_CLIENT_GOT_REJ,
     TR_PEER_PEER_GOT_DATA,
     TR_PEER_PEER_PROGRESS,
     TR_PEER_ERROR,
-    TR_PEER_CANCEL,
-    TR_PEER_UPLOAD_ONLY,
-    TR_PEER_NEED_REQ
+    TR_PEER_UPLOAD_ONLY
 }
 PeerEventType;

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 );
 

trunk/libtransmission/peer-mgr.h

@@ -26 +26 @@
 #include "bitfield.h"
 #include "net.h"
+#include "peer-common.h" /* struct peer_request */
 #include "publish.h" /* tr_publisher_tag */
 #include "utils.h"
@@ -118 +119 @@
 tr_bool tr_peerMgrPeerIsSeed( const tr_torrent * tor,
                               const tr_address * addr );
+
+void tr_peerMgrGetNextRequests( tr_torrent          * torrent,
+                                tr_peer             * peer,
+                                int                   numwant,
+                                tr_block_index_t    * setme,
+                                int                 * numgot );
+
+tr_bool tr_peerMgrDidPeerRequest( const tr_torrent  * torrent,
+                                  const tr_peer     * peer,
+                                  tr_block_index_t    block );
+
+void tr_peerMgrRebuildRequests( tr_torrent * torrent );
 
 void tr_peerMgrAddIncoming( tr_peerMgr  * manager,

trunk/libtransmission/peer-msgs.c

@@ -33 +33 @@
 #include "platform.h" /* MAX_STACK_ARRAY_SIZE */
 #include "ratecontrol.h"
-#include "request-list.h"
 #include "session.h"
 #include "stats.h"
@@ -79 +78 @@
     PEX_INTERVAL_SECS       = 90, /* sec between sendPex() calls */
 
+    REQQ                    = 512,
 
     MAX_BLOCK_SIZE          = ( 1024 * 16 ),
-
-
-    /* how long an unsent request can stay queued before it's returned
-       back to the peer-mgr's pool of requests */
-    QUEUED_REQUEST_TTL_SECS = 20,
-
-    /* how long a sent request can stay queued before it's returned
-       back to the peer-mgr's pool of requests */
-    SENT_REQUEST_TTL_SECS = 240,
 
     /* used in lowering the outMessages queue period */
     IMMEDIATE_PRIORITY_INTERVAL_SECS = 0,
     HIGH_PRIORITY_INTERVAL_SECS = 2,
-    LOW_PRIORITY_INTERVAL_SECS = 20,
+    LOW_PRIORITY_INTERVAL_SECS = 10,
 
     /* number of pieces to remove from the bitfield when
@@ -111 +102 @@
     AWAITING_BT_PIECE
 };
+
+/**
+***
+**/
+
+struct peer_request
+{
+    uint32_t    index;
+    uint32_t    offset;
+    uint32_t    length;
+};
+
+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 );
+}
+
+static void
+blockToReq( const tr_torrent     * tor,
+            tr_block_index_t       block,
+            struct peer_request  * setme )
+{
+    assert( setme != NULL );
+
+    setme->index = tr_torBlockPiece( tor, block );
+    setme->offset = getBlockOffsetInPiece( tor, block );
+    setme->length = tr_torBlockCountBytes( tor, block );
+}
 
 /**
@@ -147 +170 @@
     /*tr_bool         haveFastSet;*/
 
+    int             activeRequestCount;
+    int             desiredRequestCount;
+
     /* how long the outMessages batch should be allowed to grow before
      * it's flushed -- some messages (like requests >:) should be sent
@@ -156 +182 @@
     uint16_t        pexCount;
     uint16_t        pexCount6;
-    uint16_t        maxActiveRequests;
 
 #if 0
@@ -171 +196 @@
     struct evbuffer *      outMessages; /* all the non-piece messages */
 
-    struct request_list    peerAskedFor;
-    struct request_list    clientAskedFor;
-    struct request_list    clientWillAskFor;
-
+    struct peer_request    peerAskedFor[REQQ];
+    int                    peerAskedForCount;
+    
     tr_pex               * pex;
     tr_pex               * pex6;
@@ -443 +467 @@
 
 static void
-fireNeedReq( tr_peermsgs * msgs )
-{
-    tr_peer_event e = blankEvent;
-    e.eventType = TR_PEER_NEED_REQ;
-    publish( msgs, &e );
-}
-
-static void
 firePeerProgress( tr_peermsgs * msgs )
 {
@@ -471 +487 @@
 
 static void
+fireGotRej( tr_peermsgs * msgs, const struct peer_request * req )
+{
+    tr_peer_event e = blankEvent;
+    e.eventType = TR_PEER_CLIENT_GOT_REJ;
+    e.pieceIndex = req->index;
+    e.offset = req->offset;
+    e.length = req->length;
+    publish( msgs, &e );
+}
+
+static void
+fireGotChoke( tr_peermsgs * msgs )
+{
+    tr_peer_event e = blankEvent;
+    e.eventType = TR_PEER_CLIENT_GOT_CHOKE;
+    publish( msgs, &e );
+}
+
+static void
 fireClientGotData( tr_peermsgs * msgs,
                    uint32_t      length,
@@ -521 +556 @@
     e.wasPieceData = wasPieceData;
 
-    publish( msgs, &e );
-}
-
-static void
-fireCancelledReq( tr_peermsgs * msgs, const struct peer_request * req )
-{
-    tr_peer_event e = blankEvent;
-    e.eventType = TR_PEER_CANCEL;
-    e.pieceIndex = req->index;
-    e.offset = req->offset;
-    e.length = req->length;
     publish( msgs, &e );
 }
@@ -689 +713 @@
     if( i != msgs->peer->clientIsInterested )
         sendInterest( msgs, i );
-    if( i )
-        fireNeedReq( msgs );
 }
 
 static tr_bool
-popNextRequest( tr_peermsgs *         msgs,
-                struct peer_request * setme )
-{
-    return reqListPop( &msgs->peerAskedFor, setme );
+popNextRequest( tr_peermsgs * msgs, struct peer_request * setme )
+{
+    if( msgs->peerAskedForCount == 0 )
+        return FALSE;
+
+    *setme = msgs->peerAskedFor[0];
+    memmove( msgs->peerAskedFor, msgs->peerAskedFor + 1, --msgs->peerAskedForCount );
+    return TRUE;
 }
 
@@ -706 +732 @@
     const int mustSendCancel = tr_peerIoSupportsFEXT( msgs->peer->io );
 
-    while( popNextRequest( msgs, &req ) )
+    while( popNextRequest( msgs, &req ))
         if( mustSendCancel )
             protocolSendReject( msgs, &req );
@@ -769 +795 @@
 }
 
-static void
-expireFromList( tr_peermsgs          * msgs,
-                struct request_list  * list,
-                const time_t           oldestAllowed )
-{
-    size_t i;
-    struct request_list tmp = REQUEST_LIST_INIT;
-
-    /* since the fifo list is sorted by time, the oldest will be first */
-    if( !list->len || ( list->fifo[0].time_requested >= oldestAllowed ) )
-        return;
-
-    /* if we found one too old, start pruning them */
-    reqListCopy( &tmp, list );
-    for( i=0; i<tmp.len; ++i ) {
-        const struct peer_request * req = &tmp.fifo[i];
-        if( req->time_requested >= oldestAllowed )
-            break;
-        tr_peerMsgsCancel( msgs, req->index, req->offset, req->length );
-    }
-    reqListClear( &tmp );
-}
-
-static void
-expireOldRequests( tr_peermsgs * msgs, const time_t now  )
-{
-    time_t oldestAllowed;
-    const tr_bool fext = tr_peerIoSupportsFEXT( msgs->peer->io );
-    dbgmsg( msgs, "entering `expire old requests' block" );
-
-    /* cancel requests that have been queued for too long */
-    oldestAllowed = now - QUEUED_REQUEST_TTL_SECS;
-    expireFromList( msgs, &msgs->clientWillAskFor, oldestAllowed );
-
-    /* if the peer doesn't support "Reject Request",
-     * cancel requests that were sent too long ago. */
-    if( !fext ) {
-        oldestAllowed = now - SENT_REQUEST_TTL_SECS;
-        expireFromList( msgs, &msgs->clientAskedFor, oldestAllowed );
-    }
-
-    dbgmsg( msgs, "leaving `expire old requests' block" );
-}
-
-static void
-pumpRequestQueue( tr_peermsgs * msgs, const time_t now )
-{
-    const int           max = msgs->maxActiveRequests;
-    int                 sent = 0;
-    int                 len = msgs->clientAskedFor.len;
+
+void
+tr_peerMsgsCancel( tr_peermsgs * msgs, tr_block_index_t block )
+{
     struct peer_request req;
-
-    dbgmsg( msgs, "clientIsChoked %d, download allowed %d, len %d, max %d, msgs->clientWillAskFor.len %d",
-            (int)msgs->peer->clientIsChoked,
-            (int)tr_torrentIsPieceTransferAllowed( msgs->torrent, TR_PEER_TO_CLIENT ),
-            len, max, msgs->clientWillAskFor.len );
-
-    if( msgs->peer->clientIsChoked )
-        return;
-    if( !tr_torrentIsPieceTransferAllowed( msgs->torrent, TR_PEER_TO_CLIENT ) )
-        return;
-
-    while( ( len < max ) && reqListPop( &msgs->clientWillAskFor, &req ) )
-    {
-        const tr_block_index_t block = _tr_block( msgs->torrent, req.index, req.offset );
-
-        assert( requestIsValid( msgs, &req ) );
-        assert( tr_bitfieldHas( msgs->peer->have, req.index ) );
-
-        /* don't ask for it if we've already got it... this block may have
-         * come in from a different peer after we cancelled a request for it */
-        if( !tr_cpBlockIsComplete( &msgs->torrent->completion, block ) )
-        {
-            protocolSendRequest( msgs, &req );
-            req.time_requested = now;
-            reqListAppend( &msgs->clientAskedFor, &req );
-
-            ++len;
-            ++sent;
-        }
-        else dbgmsg( msgs, "not asking for it because we've already got it..." );
-    }
-
-    if( sent )
-        dbgmsg( msgs, "pump sent %d requests, now have %d active and %d queued",
-                sent, msgs->clientAskedFor.len, msgs->clientWillAskFor.len );
-
-    if( len < max )
-        fireNeedReq( msgs );
-}
-
-static TR_INLINE tr_bool
-requestQueueIsFull( const tr_peermsgs * msgs )
-{
-    const int req_max = msgs->maxActiveRequests;
-    return msgs->clientWillAskFor.len >= (size_t)req_max;
-}
-
-tr_addreq_t
-tr_peerMsgsAddRequest( tr_peermsgs *    msgs,
-                       uint32_t         index,
-                       uint32_t         offset,
-                       uint32_t         length )
-{
-    struct peer_request req;
-
-    assert( msgs );
-    assert( msgs->torrent );
-
-    /**
-    ***  Reasons to decline the request
-    **/
-
-    /* don't send requests to choked clients */
-    if( msgs->peer->clientIsChoked ) {
-        dbgmsg( msgs, "declining request because they're choking us" );
-        return TR_ADDREQ_CLIENT_CHOKED;
-    }
-
-    /* peer's queue is full */
-    if( requestQueueIsFull( msgs ) ) {
-        dbgmsg( msgs, "declining request because we're full" );
-        return TR_ADDREQ_FULL;
-    }
-
-    /* peer doesn't have this piece */
-    if( !tr_bitfieldHas( msgs->peer->have, index ) )
-        return TR_ADDREQ_MISSING;
-
-    /* have we already asked for this piece? */
-    req.index = index;
-    req.offset = offset;
-    req.length = length;
-    if( reqListHas( &msgs->clientAskedFor, &req ) ) {
-        dbgmsg( msgs, "declining because it's a duplicate" );
-        return TR_ADDREQ_DUPLICATE;
-    }
-    if( reqListHas( &msgs->clientWillAskFor, &req ) ) {
-        dbgmsg( msgs, "declining because it's a duplicate" );
-        return TR_ADDREQ_DUPLICATE;
-    }
-
-    /**
-    ***  Accept this request
-    **/
-
-    dbgmsg( msgs, "adding req for %"PRIu32":%"PRIu32"->%"PRIu32" to our `will request' list",
-            index, offset, length );
-    req.time_requested = time( NULL );
-    reqListAppend( &msgs->clientWillAskFor, &req );
-    return TR_ADDREQ_OK;
-}
-
-static void
-cancelAllRequestsToPeer( tr_peermsgs * msgs, tr_bool sendCancel )
-{
-    size_t i;
-    struct request_list a = msgs->clientWillAskFor;
-    struct request_list b = msgs->clientAskedFor;
-    dbgmsg( msgs, "cancelling all requests to peer" );
-
-    msgs->clientAskedFor = REQUEST_LIST_INIT;
-    msgs->clientWillAskFor = REQUEST_LIST_INIT;
-
-    for( i=0; i<a.len; ++i )
-        fireCancelledReq( msgs, &a.fifo[i] );
-
-    for( i = 0; i < b.len; ++i ) {
-        fireCancelledReq( msgs, &b.fifo[i] );
-        if( sendCancel )
-            protocolSendCancel( msgs, &b.fifo[i] );
-    }
-
-    reqListClear( &a );
-    reqListClear( &b );
-}
-
-void
-tr_peerMsgsCancel( tr_peermsgs * msgs,
-                   uint32_t      pieceIndex,
-                   uint32_t      offset,
-                   uint32_t      length )
-{
-    struct peer_request req;
-
-    assert( msgs != NULL );
-    assert( length > 0 );
-
-
-    /* have we asked the peer for this piece? */
-    req.index = pieceIndex;
-    req.offset = offset;
-    req.length = length;
-
-    /* if it's only in the queue and hasn't been sent yet, free it */
-    if( reqListRemove( &msgs->clientWillAskFor, &req ) ) {
-        dbgmsg( msgs, "cancelling %"PRIu32":%"PRIu32"->%"PRIu32, pieceIndex, offset, length );
-        fireCancelledReq( msgs, &req );
-    }
-
-    /* if it's already been sent, send a cancel message too */
-    if( reqListRemove( &msgs->clientAskedFor, &req ) ) {
-        dbgmsg( msgs, "cancelling %"PRIu32":%"PRIu32"->%"PRIu32, pieceIndex, offset, length );
-        protocolSendCancel( msgs, &req );
-        fireCancelledReq( msgs, &req );
-    }
-}
-
+    blockToReq( msgs->torrent, block, &req );
+    protocolSendCancel( msgs, &req );
+}
 
 /**
@@ -1008 +834 @@
     tr_bencDictAddInt( &val, "e", getSession(msgs)->encryptionMode != TR_CLEAR_PREFERRED );
     tr_bencDictAddInt( &val, "p", tr_sessionGetPeerPort( getSession(msgs) ) );
+    tr_bencDictAddInt( &val, "reqq", REQQ ); 
     tr_bencDictAddInt( &val, "upload_only", tr_torrentIsSeed( msgs->torrent ) );
     tr_bencDictAddStr( &val, "v", TR_NAME " " USERAGENT_PREFIX );
@@ -1252 +1079 @@
     else if( peerIsChoked )
         dbgmsg( msgs, "rejecting request from choked peer" );
+    else if( msgs->peerAskedForCount + 1 >= REQQ )
+        dbgmsg( msgs, "rejecting request ... reqq is full" );
     else
         allow = TRUE;
 
     if( allow )
-        reqListAppend( &msgs->peerAskedFor, req );
+        msgs->peerAskedFor[msgs->peerAskedForCount++] = *req;
     else if( fext )
         protocolSendReject( msgs, req );
@@ -1365 +1194 @@
 }
 
+static void updateDesiredRequestCount( tr_peermsgs * msgs, uint64_t now );
+
+static void
+decrementActiveRequestCount( tr_peermsgs * msgs )
+{
+    if( msgs->activeRequestCount > 0 )
+        msgs->activeRequestCount--;
+}
+
 static int
 readBtMessage( tr_peermsgs * msgs, struct evbuffer * inbuf, size_t inlen )
@@ -1394 +1232 @@
             msgs->peer->clientIsChoked = 1;
             if( !fext )
-                cancelAllRequestsToPeer( msgs, FALSE );
+                fireGotChoke( msgs );
             break;
 
@@ -1400 +1238 @@
             dbgmsg( msgs, "got Unchoke" );
             msgs->peer->clientIsChoked = 0;
-            fireNeedReq( msgs );
+            updateDesiredRequestCount( msgs, tr_date( ) );
             break;
 
@@ -1426 +1264 @@
 
         case BT_BITFIELD:
-        {
             dbgmsg( msgs, "got a bitfield" );
             tr_peerIoReadBytes( msgs->peer->io, inbuf, msgs->peer->have->bits, msglen );
             updatePeerProgress( msgs );
-            fireNeedReq( msgs );
             break;
-        }
 
         case BT_REQUEST:
@@ -1447 +1282 @@
         case BT_CANCEL:
         {
+            int i;
             struct peer_request r;
             tr_peerIoReadUint32( msgs->peer->io, inbuf, &r.index );
@@ -1452 +1288 @@
             tr_peerIoReadUint32( msgs->peer->io, inbuf, &r.length );
             dbgmsg( msgs, "got a Cancel %u:%u->%u", r.index, r.offset, r.length );
-            if( reqListRemove( &msgs->peerAskedFor, &r ) && fext )
-                protocolSendReject( msgs, &r );
+            for( i=0; i<msgs->peerAskedForCount; ++i ) {
+                const struct peer_request * req = msgs->peerAskedFor + i;
+                if( ( req->index == r.index ) && ( req->offset == r.offset ) && ( req->length == r.length ) )
+                    break;
+            }
+            if( i < msgs->peerAskedForCount )
+                memmove( msgs->peerAskedFor+i, msgs->peerAskedFor+i+1, --msgs->peerAskedForCount-i );
             break;
         }
@@ -1521 +1362 @@
             tr_peerIoReadUint32( msgs->peer->io, inbuf, &r.offset );
             tr_peerIoReadUint32( msgs->peer->io, inbuf, &r.length );
-            if( fext )
-                reqListRemove( &msgs->clientAskedFor, &r );
-            else {
+            if( fext ) {
+                decrementActiveRequestCount( msgs );
+                fireGotRej( msgs, &r );
+            } else {
                 fireError( msgs, EMSGSIZE );
                 return READ_ERR;
@@ -1592 +1434 @@
     dbgmsg( msgs, "got block %u:%u->%u", req->index, req->offset, req->length );
 
-    /**
-    *** Remove the block from our `we asked for this' list
-    **/
-
-    if( !reqListRemove( &msgs->clientAskedFor, req ) ) {
-        clientGotUnwantedBlock( msgs, req );
+    if( !tr_peerMgrDidPeerRequest( msgs->torrent, msgs->peer, block ) ) {
         dbgmsg( msgs, "we didn't ask for this message..." );
-        return 0;
-    }
-
-    dbgmsg( msgs, "peer has %d more blocks we've asked for",
-            msgs->clientAskedFor.len );
-
-    /**
-    *** Error checks
-    **/
-
-    if( tr_cpBlockIsComplete( &tor->completion, block ) ) {
-        dbgmsg( msgs, "we have this block already..." );
-        clientGotUnwantedBlock( msgs, req );
         return 0;
     }
@@ -1623 +1447 @@
 
     addPeerToBlamefield( msgs, req->index );
+    decrementActiveRequestCount( msgs );
     fireGotBlock( msgs, req );
     return 0;
@@ -1686 +1511 @@
 **/
 
-static int
-ratePulse( tr_peermsgs * msgs, uint64_t now )
-{
-    int irate;
-    const int floor = 8;
-    const int seconds = 10;
-    double rate;
-    int estimatedBlocksInPeriod;
+static void
+updateDesiredRequestCount( tr_peermsgs * msgs, uint64_t now )
+{
     const tr_torrent * const torrent = msgs->torrent;
 
-    /* Get the rate limit we should use.
-     * FIXME: this needs to consider all the other peers as well... */
-    rate = tr_peerGetPieceSpeed( msgs->peer, now, TR_PEER_TO_CLIENT );
-    if( tr_torrentUsesSpeedLimit( torrent, TR_PEER_TO_CLIENT ) )
-        rate = MIN( rate, tr_torrentGetSpeedLimit( torrent, TR_PEER_TO_CLIENT ) );
-    if( tr_torrentUsesSessionLimits( torrent ) )
-        if( tr_sessionGetActiveSpeedLimit( torrent->session, TR_PEER_TO_CLIENT, &irate ) )
-            rate = MIN( rate, irate );
-
-    estimatedBlocksInPeriod = ( rate * seconds * 1024 ) / torrent->blockSize;
-    msgs->maxActiveRequests = MAX( floor, estimatedBlocksInPeriod );
-
-    if( msgs->reqq > 0 )
-        msgs->maxActiveRequests = MIN( msgs->maxActiveRequests, msgs->reqq );
-
-    return TRUE;
+    if( tr_torrentIsSeed( msgs->torrent ) )
+    {
+        msgs->desiredRequestCount = 0;
+    }
+    else if( msgs->peer->clientIsChoked )
+    {
+        msgs->desiredRequestCount = 0;
+    }
+    else
+    {
+        int irate;
+        int estimatedBlocksInPeriod;
+        double rate;
+        const int floor = 16;
+        const int seconds = REQUEST_BUF_SECS; 
+
+        /* Get the rate limit we should use.
+         * FIXME: this needs to consider all the other peers as well... */
+        rate = tr_peerGetPieceSpeed( msgs->peer, now, TR_PEER_TO_CLIENT );
+        if( tr_torrentUsesSpeedLimit( torrent, TR_PEER_TO_CLIENT ) )
+            rate = MIN( rate, tr_torrentGetSpeedLimit( torrent, TR_PEER_TO_CLIENT ) );
+
+        /* honor the session limits, if enabled */
+        if( tr_torrentUsesSessionLimits( torrent ) )
+            if( tr_sessionGetActiveSpeedLimit( torrent->session, TR_PEER_TO_CLIENT, &irate ) )
+                rate = MIN( rate, irate );
+
+        /* use this desired rate to figure out how
+         * many requests we should send to this peer */
+        estimatedBlocksInPeriod = ( rate * seconds * 1024 ) / torrent->blockSize;
+        msgs->desiredRequestCount = MAX( floor, estimatedBlocksInPeriod );
+
+        /* honor the peer's maximum request count, if specified */
+        if( msgs->reqq > 0 )
+            if( msgs->desiredRequestCount > msgs->reqq )
+                msgs->desiredRequestCount = msgs->reqq;
+    }
+}
+
+static void
+updateRequests( tr_peermsgs * msgs )
+{
+    const int MIN_BATCH_SIZE = 4;
+    const int numwant = msgs->desiredRequestCount - msgs->activeRequestCount;
+
+    /* make sure we have enough block requests queued up */
+    if( numwant >= MIN_BATCH_SIZE )
+    {
+        int i;
+        int n;
+        tr_block_index_t * blocks = tr_new( tr_block_index_t, numwant );
+
+        tr_peerMgrGetNextRequests( msgs->torrent, msgs->peer, numwant, blocks, &n ); 
+
+        for( i=0; i<n; ++i )
+        {
+            struct peer_request req;
+            blockToReq( msgs->torrent, blocks[i], &req );
+            protocolSendRequest( msgs, &req );
+        }
+
+        msgs->activeRequestCount += n;
+
+        tr_free( blocks );
+    }
 }
 
@@ -1819 +1688 @@
     const time_t  now = time( NULL );
 
-    ratePulse( msgs, now );
-
-    pumpRequestQueue( msgs, now );
-    expireOldRequests( msgs, now );
+    if ( tr_isPeerIo( msgs->peer->io ) ) {
+        updateDesiredRequestCount( msgs, now );
+        updateRequests( msgs );
+    }
 
     for( ;; )
@@ -2180 +2049 @@
     m->outMessagesBatchPeriod = LOW_PRIORITY_INTERVAL_SECS;
     m->incoming.block = evbuffer_new( );
-    m->peerAskedFor = REQUEST_LIST_INIT;
-    m->clientAskedFor = REQUEST_LIST_INIT;
-    m->clientWillAskFor = REQUEST_LIST_INIT;
+    m->peerAskedForCount = 0;
     evtimer_set( &m->pexTimer, pexPulse, m );
     tr_timerAdd( &m->pexTimer, PEX_INTERVAL_SECS, 0 );
@@ -2198 +2065 @@
 
     tr_peerIoSetIOFuncs( m->peer->io, canRead, didWrite, gotError, m );
-    ratePulse( m, tr_date() );
+    updateDesiredRequestCount( m, tr_date( ) );
 
     return m;
@@ -2210 +2077 @@
         evtimer_del( &msgs->pexTimer );
         tr_publisherDestruct( &msgs->publisher );
-        reqListClear( &msgs->clientWillAskFor );
-        reqListClear( &msgs->clientAskedFor );
-        reqListClear( &msgs->peerAskedFor );
 
         evbuffer_free( msgs->incoming.block );

trunk/libtransmission/peer-msgs.h

@@ -49 +49 @@
 
 void         tr_peerMsgsCancel( tr_peermsgs * msgs,
-                                uint32_t      pieceIndex,
-                                uint32_t      offset,
-                                uint32_t      length );
-
+                                tr_block_index_t block );
 
 void         tr_peerMsgsFree( tr_peermsgs* );
-
-tr_addreq_t  tr_peerMsgsAddRequest( tr_peermsgs * peer,
-                                    uint32_t      pieceIndex,
-                                    uint32_t      offset,
-                                    uint32_t      length );
 
 void         tr_peerMsgsUnsubscribe( tr_peermsgs      * peer,

trunk/libtransmission/torrent.c

@@ -1678 +1678 @@
 {
     tr_file_index_t i;
-
-    assert( tr_isTorrent( tor ) );
-
+    assert( tr_isTorrent( tor ) );
     tr_torrentLock( tor );
 
     for( i = 0; i < fileCount; ++i )
         tr_torrentInitFilePriority( tor, files[i], priority );
-
     tr_torrentSetDirty( tor );
+    tr_peerMgrRebuildRequests( tor );
+
     tr_torrentUnlock( tor );
 }
@@ -1827 +1826 @@
 {
     assert( tr_isTorrent( tor ) );
-
     tr_torrentLock( tor );
+
     tr_torrentInitFileDLs( tor, files, fileCount, doDownload );
     tr_torrentSetDirty( tor );
+    tr_peerMgrRebuildRequests( tor );
+
     tr_torrentUnlock( tor );
 }

trunk/libtransmission/webseed.c

@@ -65 +65 @@
 fireNeedReq( tr_webseed * w )
 {
+#if 0
     tr_peer_event e = blankEvent;
     e.eventType = TR_PEER_NEED_REQ;
     publish( w, &e );
+#endif
 }