Changeset 3227

Timestamp:

Sep 28, 2007, 4:40:21 PM (15 years ago)

Author:

charles

Message:

commit the next step of tiennou's fastpeers code.

Location:

trunk/libtransmission

Files:

• handshake.c (modified) (8 diffs)
• peer-io.c (modified) (1 diff)
• peer-mgr.c (modified) (6 diffs)
• peer-mgr.h (modified) (1 diff)
• peer-msgs.c (modified) (9 diffs)

trunk/libtransmission/handshake.c

@@ -37 +37 @@
 
 /* enable fast peers extension protocol */
-/* (disabled because it's not fully implemented yet) */
 /* #define ENABLE_FASTPEER */
 
@@ -240 +239 @@
     }
     
-    HANDSHAKE_SET_FASTEXT ( walk );
+    HANDSHAKE_SET_FASTEXT( walk );
 
     walk += HANDSHAKE_FLAGS_LEN;
@@ -331 +330 @@
     assert( !ltep || !azmp );
     
-         if( ltep ) { tr_peerIoEnableLTEP( handshake->io, 1 ); dbgmsg(handshake,"using ltep" ); }
-    else if( azmp ) { tr_peerIoEnableAZMP( handshake->io, 1 ); dbgmsg(handshake,"using azmp" ); }
-    else if( fpex ) { tr_peerIoEnableFEXT( handshake->io, 1 ); dbgmsg(handshake,"using fext" ); }
-    else            { dbgmsg(handshake,"using no extensions" ); }
-
+    if( ltep ) { tr_peerIoEnableLTEP( handshake->io, 1 ); dbgmsg(handshake,"using ltep" ); }
+    if( azmp ) { tr_peerIoEnableAZMP( handshake->io, 1 ); dbgmsg(handshake,"using azmp" ); }
+    if( fpex ) { tr_peerIoEnableFEXT( handshake->io, 1 ); dbgmsg(handshake,"using fext" ); }
+        
     return HANDSHAKE_OK;
 }
@@ -699 +697 @@
     assert( !ltep || !azmp );
     
-         if( ltep ) { tr_peerIoEnableLTEP( handshake->io, 1 ); dbgmsg(handshake,"using ltep" ); }
-    else if( azmp ) { tr_peerIoEnableAZMP( handshake->io, 1 ); dbgmsg(handshake,"using azmp" ); }
-    else if( fpex ) { tr_peerIoEnableFEXT( handshake->io, 1 ); dbgmsg(handshake,"using fext" ); }
-    else            { dbgmsg(handshake,"using no extensions" ); }
+    if( ltep ) { tr_peerIoEnableLTEP( handshake->io, 1 ); dbgmsg(handshake,"using ltep" ); }
+    if( azmp ) { tr_peerIoEnableAZMP( handshake->io, 1 ); dbgmsg(handshake,"using azmp" ); }
+    if( fpex ) { tr_peerIoEnableFEXT( handshake->io, 1 ); dbgmsg(handshake,"using fext" ); }
     
     /**
@@ -946 +943 @@
             ext = 0;
 
-        /* FIXME: do we do something with fast peers here? */
-
         msg = buildHandshakeMessage( handshake, ext, &msgSize );
         tr_peerIoWriteBytes( handshake->io, outbuf, msg, msgSize );
@@ -1030 +1025 @@
 {
     tr_handshake * handshake = (tr_handshake *) arg;
-dbgmsg( handshake, "handshake evbuffer got an error!!!!!" );
 
     /* if the error happened while we were sending a public key, we might
@@ -1039 +1033 @@
         && ( !tr_peerIoReconnect( handshake->io ) ) )
     {
+        dbgmsg( handshake, "handshake failed, trying plaintext..." );
         int msgSize; 
         uint8_t * msg = buildHandshakeMessage( handshake, HANDSHAKE_EXTPREF_LTEP_PREFER, &msgSize );
@@ -1047 +1042 @@
     else
     {
-dbgmsg( handshake, "handshake evbuffer got an error!!!!!" );
+        dbgmsg( handshake, "handshake evbuffer got an error!!!!!" );
         tr_handshakeDone( handshake, FALSE );
     }

trunk/libtransmission/peer-io.c

@@ -34 +34 @@
     unsigned int extendedProtocolSupported : 1;
     unsigned int fastPeersSupported : 1;
-    /* unsigned int DHTSupported : 1; */
 };
 

trunk/libtransmission/peer-mgr.c

@@ -20 +20 @@
 #include "clients.h"
 #include "completion.h"
+#include "crypto.h"
 #include "handshake.h"
 #include "net.h"
@@ -275 +276 @@
 **/
 
+struct tr_bitfield *
+tr_peerMgrGenerateAllowedSet( const uint32_t         setCount,
+                              const uint32_t         pieceCount,
+                              const uint8_t          infohash[20],
+                              const struct in_addr * ip )
+{
+    /* This has been checked against the spec example implementation. Feeding it with :
+    setCount = 9, pieceCount = 1313, infohash = Oxaa,0xaa,...0xaa, ip = 80.4.4.200
+generate :
+    1059, 431, 808, 1217, 287, 376, 1188, 353, 508
+    but since we're storing in a bitfield, it won't be in this order... */
+    /* TODO : We should translate link-local IPv4 adresses to external IP, 
+     * so that being on same local network gives us the same allowed pieces */
+    
+    printf( "%d piece allowed fast set for torrent with %d pieces and hex infohash\n", setCount, pieceCount );
+    printf( "%x%x%x%x%x%x%x%x%x%x%x%x%x%x%x%x%x%x%x%x for node with IP %s:\n",
+            infohash[0], infohash[1], infohash[2], infohash[3], infohash[4], infohash[5], infohash[6], infohash[7], infohash[8], infohash[9],
+            infohash[10], infohash[11], infohash[12], infohash[13], infohash[14], infohash[15], infohash[16], infohash[7], infohash[18], infohash[19],
+            inet_ntoa( *ip ) );
+    
+    uint8_t *seed = malloc(4 + SHA_DIGEST_LENGTH);
+    char buf[4];
+    uint32_t allowedPieceCount = 0;
+    tr_bitfield_t * ret;
+    
+    ret = tr_bitfieldNew( pieceCount );
+    
+    /* We need a seed based on most significant bytes of peer address
+        concatenated with torrent's infohash */
+    *(uint32_t*)buf = ntohl( htonl(ip->s_addr) & 0xffffff00 );
+    
+    memcpy( seed, &buf, 4 );
+    memcpy( seed + 4, infohash, SHA_DIGEST_LENGTH );
+    
+    tr_sha1( seed, seed, 4 + SHA_DIGEST_LENGTH, NULL );
+    
+    while ( allowedPieceCount < setCount )
+    {
+        int i;
+        for ( i = 0 ; i < 5 && allowedPieceCount < setCount ; i++ )
+        {
+            /* We generate indices from 4-byte chunks of the seed */
+            uint32_t j = i * 4;
+            uint32_t y = ntohl( *(uint32_t*)(seed + j) );
+            uint32_t index = y % pieceCount;
+            
+            if ( !tr_bitfieldHas( ret, index ) )
+            {
+                tr_bitfieldAdd( ret, index );
+                allowedPieceCount++;
+            }
+        }
+        /* We randomize the seed, in case we need to iterate more */
+        tr_sha1( seed, seed, SHA_DIGEST_LENGTH, NULL );
+    }
+    tr_free( seed );
+    
+    return ret;
+}
+
 tr_peerMgr*
 tr_peerMgrNew( tr_handle * handle )
@@ -323 +384 @@
     uint32_t piece;
     uint32_t peerCount;
+    uint32_t fastAllowed;
 };
 
@@ -338 +400 @@
     if (a->peerCount != b->peerCount)
         return a->peerCount < b->peerCount ? -1 : 1;
+    
+    /* otherwise if one *might be* fastallowed to us */
+    if (a->fastAllowed != b->fastAllowed)
+        return a->fastAllowed < b->fastAllowed ? -1 : 1;
 
     /* otherwise go with the earlier piece */
@@ -389 +455 @@
             setme->priority = inf->pieces[piece].priority;
             setme->peerCount = 0;
+            /* FIXME */
+//            setme->fastAllowed = tr_bitfieldHas( t->tor->allowedList, i);
 
             for( k=0; k<peerCount; ++k ) {
@@ -1118 +1186 @@
     tr_peer ** peers = getConnectedPeers( t, &peerCount );
     ChokeData * choke = tr_new0( ChokeData, peerCount );
-
+    
     /* sort the peers by preference and rate */
     for( i=0; i<peerCount; ++i )

trunk/libtransmission/peer-mgr.h

@@ -97 +97 @@
                                            int               * setmeCount );
 
-
+                             
+struct tr_bitfield *
+tr_peerMgrGenerateAllowedSet( const uint32_t         setCount,
+                              const uint32_t         pieceCount,
+                              const uint8_t          infohash[20],
+                              const struct in_addr * ip );
 
 

trunk/libtransmission/peer-msgs.c

@@ -38 +38 @@
 ***
 **/
+
+#define MAX_ALLOWED_SET_COUNT   10 /* number of pieces generated for allow-fast,
+                                    threshold for fast-allowing others */
 
 enum
@@ -123 +126 @@
     unsigned int peerSupportsPex       : 1;
     unsigned int hasSentLtepHandshake  : 1;
-
+    
+    tr_bitfield * clientAllowedPieces;
+    tr_bitfield * peerAllowedPieces;
+    
     uint8_t state;
 
@@ -300 +306 @@
     if( msgs->info->peerIsChoked != choke )
     {
-        msgs->info->peerIsChoked = choke ? 1 : 0;
-
+        msgs->info->peerIsChoked = choke;
+        tr_list * walk;
+        
         if( choke )
-            tr_list_free( &msgs->peerAskedFor, tr_free );
+            for( walk = msgs->peerAskedFor; walk != NULL; )
+            {
+                tr_list * next = walk->next;
+                /* We shouldn't reject a peer's fast allowed requests at choke */
+                struct peer_request *req = walk->data;
+                if ( !tr_bitfieldHas( msgs->peerAllowedPieces, req->index ) )
+                {
+                    tr_list_remove_data( &msgs->peerAskedFor, req );
+                    tr_free( req );
+                }
+                walk = next;
+            }
 
         dbgmsg( msgs, "sending a %s message", (choke ? "CHOKE" : "UNCHOKE") );
@@ -360 +378 @@
     updateInterest( msgs );
 }
+#if 0
+static void
+sendFastSuggest( tr_peermsgs * msgs,
+                 uint32_t      pieceIndex )
+{
+    dbgmsg( msgs, "w00t SUGGESTing them piece #%d", pieceIndex );
+    tr_peerIoWriteUint32( msgs->io, msgs->outMessages, sizeof(uint8_t) + sizeof(uint32_t) );
+    tr_peerIoWriteUint8( msgs->io, msgs->outMessages, BT_SUGGEST );
+    tr_peerIoWriteUint32( msgs->io, msgs->outMessages, pieceIndex );
+    
+    updateInterest( msgs );
+}
+#endif
+static void
+sendFastHave( tr_peermsgs * msgs,
+              int           all)
+{
+    dbgmsg( msgs, "w00t telling them we %s pieces", (all ? "HAVE_ALL" : "HAVE_NONE" ) );
+    tr_peerIoWriteUint32( msgs->io, msgs->outMessages, sizeof(uint8_t) );
+    tr_peerIoWriteUint8( msgs->io, msgs->outMessages, ( all ? BT_HAVE_ALL : BT_HAVE_NONE ) );
+    
+    updateInterest( msgs );
+}
+
+static void
+sendFastReject( tr_peermsgs * msgs,
+                uint32_t      pieceIndex,
+                uint32_t      offset,
+                uint32_t      length )
+{
+    assert( msgs != NULL );
+    assert( length > 0 );
+    
+    /* reject the request */
+    const uint32_t len = sizeof(uint8_t) + 3 * sizeof(uint32_t);
+    tr_peerIoWriteUint32( msgs->io, msgs->outMessages, len );
+    tr_peerIoWriteUint8( msgs->io, msgs->outMessages, BT_REJECT );
+    tr_peerIoWriteUint32( msgs->io, msgs->outMessages, pieceIndex );
+    tr_peerIoWriteUint32( msgs->io, msgs->outMessages, offset );
+    tr_peerIoWriteUint32( msgs->io, msgs->outMessages, length );
+}
+
+static void
+sendFastAllowed( tr_peermsgs * msgs,
+                 uint32_t      pieceIndex)
+{
+    dbgmsg( msgs, "w00t telling them we ALLOW_FAST piece #%d", pieceIndex );
+    tr_peerIoWriteUint32( msgs->io, msgs->outMessages, sizeof(uint8_t) + sizeof(uint32_t) );
+    tr_peerIoWriteUint8( msgs->io, msgs->outMessages, BT_ALLOWED_FAST );
+    tr_peerIoWriteUint32( msgs->io, msgs->outMessages, pieceIndex );
+}
+
+
+
+static void
+sendFastAllowedSet( tr_peermsgs * msgs )
+{
+    int i = 0;
+    while (i <= msgs->torrent->info.pieceCount )
+    {
+        if ( tr_bitfieldHas( msgs->peerAllowedPieces, i) )
+            sendFastAllowed( msgs, i );
+        i++;
+    }
+}
+
 
 /**
@@ -633 +717 @@
             assert( msglen == 0 );
             msgs->info->clientIsChoked = 1;
-            tr_list_free( &msgs->peerAskedFor, tr_free );
+            
+            tr_list * walk;
+            for( walk = msgs->peerAskedFor; walk != NULL; )
+            {
+                tr_list * next = walk->next;
+                /* We shouldn't reject a peer's fast allowed requests at choke */
+                struct peer_request *req = walk->data;
+                if ( !tr_bitfieldHas( msgs->peerAllowedPieces, req->index ) )
+                {
+                    tr_list_remove_data( &msgs->peerAskedFor, req );
+                    tr_free( req );
+                }
+                walk = next;
+            }
             tr_list_free( &msgs->clientAskedFor, tr_free );
             break;
@@ -686 +783 @@
             tr_peerIoReadUint32( msgs->io, inbuf, &req->offset );
             tr_peerIoReadUint32( msgs->io, inbuf, &req->length );
-            if( !msgs->info->peerIsChoked && requestIsValid( msgs, req ) )
-                tr_list_append( &msgs->peerAskedFor, req );
-            else
+            
+            if ( !requestIsValid( msgs, req ) )
+            {
+                dbgmsg( msgs, "BT_REQUEST: invalid request, ignoring" );
                 tr_free( req );
+                break;
+            }
+            /* 
+                If we're not choking him -> continue
+                If we're choking him
+                    it doesn't support FPE -> He's deaf, reCHOKE and bail...
+                    it support FPE
+                        If the asked piece is not allowed
+                            OR he's above our threshold
+                            OR we don't have the requested piece -> Reject
+                        Else
+                        Asked piece allowed AND he's below our threshold -> continue...
+             */
+    
+
+            if ( msgs->info->peerIsChoked )
+            {
+                if ( !tr_peerIoSupportsFEXT( msgs->io ) )
+                {
+                    dbgmsg( msgs, "BT_REQUEST: peer is choked, ignoring" );
+                    /* Didn't he get it? */
+                    tr_peerMsgsSetChoke( msgs, 1 );
+                    tr_free( req );
+                    break;
+                }
+                else
+                {
+                    if ( !tr_bitfieldHas( msgs->peerAllowedPieces, req->index )
+                         || ( msgs->info->progress * (float)msgs->torrent->info.pieceCount) >= MAX_ALLOWED_SET_COUNT
+                         || !tr_cpPieceIsComplete( msgs->torrent->completion, req->index ) )
+                    {
+                        dbgmsg( msgs, "BT_REQUEST: peer requests an un-fastallowed piece" );
+                        sendFastReject( msgs, req->index, req->offset, req->length );
+                        tr_free( req );
+                        break;
+                    }
+                    dbgmsg( msgs, "BT_REQUEST: fast allowed piece, accepting request" );
+                }    
+            }
+            
+            tr_list_append( &msgs->peerAskedFor, req );
             break;
         }
@@ -732 +871 @@
             
         case BT_HAVE_ALL: {
-            /* tiennou TODO */
+            assert( msglen == 0 );
+            dbgmsg( msgs, "peer sent us a BT_HAVE_ALL" );
+            memset( msgs->info->have->bits, 1, msgs->info->have->len );
+            msgs->info->progress = tr_bitfieldCountTrueBits( msgs->info->have ) / (float)msgs->torrent->info.pieceCount;
+            dbgmsg( msgs, "after the HAVE_ALL message, peer progress is %f", msgs->info->progress );
+            updateInterest( msgs );
+            firePeerProgress( msgs );
             break;
         }
             
         case BT_HAVE_NONE: {
-            /* tiennou TODO */
+            assert( msglen == 0 );
+            dbgmsg( msgs, "peer sent us a BT_HAVE_NONE" );
+            memset( msgs->info->have->bits, 1, msgs->info->have->len );
+            msgs->info->progress = tr_bitfieldCountTrueBits( msgs->info->have ) / (float)msgs->torrent->info.pieceCount;
+            dbgmsg( msgs, "after the HAVE_NONE message, peer progress is %f", msgs->info->progress );
+            updateInterest( msgs );
+            firePeerProgress( msgs );
             break;
         }
             
         case BT_REJECT: {
-            /* tiennou TODO */
+            struct peer_request req;
+            tr_list * node;
+            assert( msglen == 12 );
+            dbgmsg( msgs, "peer sent us a BT_REJECT" );
+            tr_peerIoReadUint32( msgs->io, inbuf, &req.index );
+            tr_peerIoReadUint32( msgs->io, inbuf, &req.offset );
+            tr_peerIoReadUint32( msgs->io, inbuf, &req.length );
+            node = tr_list_find( msgs->peerAskedFor, &req, peer_request_compare );
+            if( node != NULL ) {
+                void * data = node->data;
+                tr_list_remove_data( &msgs->peerAskedFor, data );
+                tr_free( data );
+                dbgmsg( msgs, "found the req that peer has rejected... cancelled." );
+            }
             break;
         }
             
         case BT_ALLOWED_FAST: {
-            /* tiennou TODO */
+            assert( msglen == 4 );
+            dbgmsg( msgs, "peer sent us a BT_ALLOWED_FAST" );
+            tr_peerIoReadUint32( msgs->io, inbuf, &ui32 );
+            tr_bitfieldAdd( msgs->clientAllowedPieces, ui32 );
             break;
         }
@@ -1284 +1451 @@
     msgs->outBlock = evbuffer_new( );
     msgs->inBlock = evbuffer_new( );
-
+    msgs->peerAllowedPieces = NULL;
+    msgs->clientAllowedPieces = NULL;
+    
+    if ( tr_peerIoSupportsFEXT( msgs->io ) )
+    {
+        /* This peer is fastpeer-enabled, generate its allowed set
+         * (before registering our callbacks) */
+        if ( !msgs->peerAllowedPieces ) {
+            const struct in_addr *peerAddr = tr_peerIoGetAddress( msgs->io, NULL );
+            
+            msgs->peerAllowedPieces = tr_peerMgrGenerateAllowedSet( MAX_ALLOWED_SET_COUNT,
+                                                                    msgs->torrent->info.pieceCount,
+                                                                    msgs->torrent->info.hash,
+                                                                    peerAddr );
+        }
+        msgs->clientAllowedPieces = tr_bitfieldNew( msgs->torrent->info.pieceCount );
+    }
+    
     tr_peerIoSetIOFuncs( msgs->io, canRead, didWrite, gotError, msgs );
     tr_peerIoSetIOMode( msgs->io, EV_READ|EV_WRITE, 0 );
@@ -1305 +1489 @@
         }
     }
-
-    sendBitfield( msgs );
+    
+    if ( tr_peerIoSupportsFEXT( msgs->io ) )
+    {
+        /* This peer is fastpeer-enabled, send it have-all or have-none if appropriate */
+        float completion = tr_cpPercentComplete( msgs->torrent->completion );
+        if ( completion == 0.0f ) {
+            sendFastHave( msgs, 0 );
+        } else if ( completion == 1.0f ) {
+            sendFastHave( msgs, 1 );
+        } else {
+            sendBitfield( msgs );
+        }
+        uint32_t peerProgress = msgs->torrent->info.pieceCount * msgs->info->progress;
+        
+        if ( peerProgress < MAX_ALLOWED_SET_COUNT )
+            sendFastAllowedSet( msgs );
+    } else {
+        sendBitfield( msgs );
+    }
     return msgs;
 }