Changeset 3065

Timestamp:

Sep 15, 2007, 12:39:55 AM (16 years ago)

Author:

charles

Message:

handle peers sending us corrupted blocks. better handling and broadcasting of `have' messages.

Location:

branches/encryption/libtransmission

Files:

• handshake.c (modified) (3 diffs)
• inout.c (modified) (1 diff)
• inout.h (modified) (1 diff)
• peer-mgr.c (modified) (13 diffs)
• peer-msgs.c (modified) (25 diffs)
• peer-msgs.h (modified) (1 diff)
• shared.c (modified) (1 diff)

branches/encryption/libtransmission/handshake.c

@@ -625 +625 @@
     return READ_AGAIN;
 }
+
+static void
+gotError( struct bufferevent * evbuf UNUSED, short what UNUSED, void * arg );
 
 /*ccc*/
@@ -676 +679 @@
     fprintf( stderr, "pstrlen is [%s]\n", pstr );
     bytesRead += pstrlen;
-    assert( !strcmp( (char*)pstr, "BitTorrent protocol" ) );
+    fprintf( stderr, "%*.*s", pstrlen, pstrlen, (char*)pstr );
+    if( strcmp( (char*)pstr, "BitTorrent protocol" ) ) {
+        tr_free( pstr );
+        gotError( NULL, 0, handshake );
+        return READ_DONE;
+    }
     tr_free( pstr );
 
@@ -796 +804 @@
 
 static void
-gotError( struct bufferevent * evbuf UNUSED, short what, void * arg )
+gotError( struct bufferevent * evbuf UNUSED, short what UNUSED, void * arg )
 {
     tr_handshake * handshake = (tr_handshake *) arg;
-fprintf( stderr, "handshake %p: got error [%s]; what==%hd... state was [%s]\n", handshake, strerror(errno), what, getStateName(handshake->state) );
 
     /* if the error happened while we were sending a public key, we might

branches/encryption/libtransmission/inout.c

@@ -252 +252 @@
 
 int
-tr_ioHash( tr_io * io, int pieceIndex )
+tr_ioHash( tr_torrent * tor, int pieceIndex )
 {
     int ret;
-    tr_torrent * tor = io->tor;
     const int success = !checkPiece( tor, pieceIndex );
 

branches/encryption/libtransmission/inout.h

@@ -41 +41 @@
 int tr_ioWrite ( struct tr_torrent *, int index, int begin, int len, uint8_t * );
 
-/***********************************************************************
- * tr_ioHash
- ***********************************************************************
- * Hashes the specified piece and updates the completion accordingly.
- **********************************************************************/
-int tr_ioHash ( tr_io *, int piece );
+/* hashes the specified piece and updates the completion accordingly. */
+int tr_ioHash ( tr_torrent*, int piece );
 
 /* close all the files associated with this torrent*/

branches/encryption/libtransmission/peer-mgr.c

@@ -32 +32 @@
 #define RECHOKE_PERIOD_SECONDS (15 * 1000)
 
-/* how many downloaders to unchoke per-torrent.
- * http://wiki.theory.org/BitTorrentSpecification#Choking_and_Optimistic_Unchoking */
-#define NUM_DOWNLOADERS_TO_UNCHOKE 6
+/* how many peers to unchoke per-torrent. */
+/* FIXME: make this user-configurable */
+#define NUM_UNCHOKED_PEERS_PER_TORRENT 8
 
 /**
@@ -400 +400 @@
                 const int val = tr_peerMsgsAddRequest( peers[j]->msgs, index, begin, length );
                 if( val==TR_ADDREQ_FULL || val==TR_ADDREQ_CLIENT_CHOKED ) {
-                    fprintf( stderr, "peer %p (of %d) is full\n", peers[j]->msgs, size );
+                    fprintf( stderr, "peer %p (of %d) is full or is choking us\n", peers[j]->msgs, size );
                     peers[j] = peers[--size];
                 }
@@ -409 +409 @@
                     fprintf( stderr, "peer %p took the request for block %d\n", peers[j]->msgs, b );
                     incrementReqCount( &t->blocks[i] );
-                    j = size;
+                    break;
                 }
             }
@@ -430 +430 @@
 {
     if( t->refillTimer == NULL ) {
-        t->refillTimer = tr_timerNew( t->manager->handle, refillPulse, t, 750 );
+        t->refillTimer = tr_timerNew( t->manager->handle, refillPulse, t, 250 );
         fprintf( stderr, "timer REFILL %p is new\n", t->refillTimer );
     }
@@ -436 +436 @@
 
 static void
-broadcastHave( Torrent * t, uint32_t index )
+broadcastClientHave( Torrent * t, uint32_t index )
 {
     int i, size;
@@ -463 +463 @@
     switch( e->eventType )
     {
-        case TR_PEERMSG_GOT_BITFIELD: {
+        case TR_PEERMSG_PEER_BITFIELD: {
             if( t->blocks!=NULL ) {
                 int i;
@@ -479 +479 @@
         }
 
-        case TR_PEERMSG_GOT_HAVE: {
-            const uint32_t begin = tr_torPieceFirstBlock( t->tor, e->pieceIndex );
-            const uint32_t end = begin + tr_torPieceCountBlocks( t->tor, (int)e->pieceIndex );
-            uint32_t i;
-            for( i=begin; t->blocks!=NULL && i<end; ++i ) {
-                assert( t->blocks[i].block == i );
-                incrementScarcity( &t->blocks[i] );
+        case TR_PEERMSG_CLIENT_HAVE:
+            broadcastClientHave( t, e->pieceIndex );
+            break;
+
+        case TR_PEERMSG_PEER_HAVE: {
+            if( t->blocks != NULL ) {
+                uint32_t i;
+                const uint32_t begin = tr_torPieceFirstBlock( t->tor, e->pieceIndex );
+                const uint32_t end = begin + tr_torPieceCountBlocks( t->tor, (int)e->pieceIndex );
+                for( i=begin; i<end; ++i ) {
+                    assert( t->blocks[i].block == i );
+                    incrementScarcity( &t->blocks[i] );
+                }
             }
-            broadcastHave( t, e->pieceIndex );
             break;
         }
 
-        case TR_PEERMSG_GOT_BLOCK: {
+        case TR_PEERMSG_CLIENT_BLOCK: {
             if( t->blocks != NULL ) {
                 const uint32_t i = _tr_block( t->tor, e->pieceIndex, e->offset );
@@ -697 +702 @@
                     int              success UNUSED )
 {
-    assert( 0 );
+    fprintf( stderr, "FIXME: tr_peerMgrSetBlame\n" );
 }
 
@@ -941 +946 @@
     tr_peer * peer;
     float rate;
-    int isInterested;
+    int randomKey;
 }
 ChokeData;
 
 static int
-compareChokeByRate( const void * va, const void * vb )
+compareChoke( const void * va, const void * vb )
 {
     const ChokeData * a = ( const ChokeData * ) va;
@@ -952 +957 @@
     if( a->rate > b->rate ) return -1;
     if( a->rate < b->rate ) return 1;
-    return 0;
-}
-
-static int
-compareChokeByDownloader( const void * va, const void * vb )
-{
-    const ChokeData * a = ( const ChokeData * ) va;
-    const ChokeData * b = ( const ChokeData * ) vb;
-
-    /* primary key: interest */
-    if(  a->isInterested && !b->isInterested ) return -1;
-    if( !a->isInterested &&  b->isInterested ) return 1;
-
-    /* second key: rate */
-    return compareChokeByRate( va, vb );
+    return a->randomKey - b->randomKey;
 }
 
@@ -975 +966 @@
     int i, size, unchoked;
     const int done = tr_cpGetStatus( t->tor->completion ) != TR_CP_INCOMPLETE;
-    float bestDownloaderRate;
     ChokeData * data;
     tr_peer ** peers = getConnectedPeers( t, &size );
     const time_t now = time( NULL );
+    int optimistic = FALSE;
 
 fprintf( stderr, "[%s] rechoking torrent %p, with %d peers\n", ctime(&now), t, size );
@@ -990 +981 @@
     for( i=0; i<size; ++i ) {
         data[i].peer = peers[i];
-        data[i].isInterested = peers[i]->peerIsInterested;
+        data[i].randomKey = tr_rand( INT_MAX );
         data[i].rate = done ? tr_peerIoGetRateToPeer( peers[i]->io )
                             : tr_peerIoGetRateToClient( peers[i]->io );
     }
 
-    /* find the best downloaders and unchoke them */
-    qsort( data, size, sizeof(ChokeData), compareChokeByDownloader );
-    bestDownloaderRate = data[0].rate;
-    for( i=unchoked=0; i<size && unchoked<NUM_DOWNLOADERS_TO_UNCHOKE; ++i ) {
+    /* find the best peers and unchoke them */
+    qsort( data, size, sizeof(ChokeData), compareChoke );
+    for( i=unchoked=0; i<size && unchoked<NUM_UNCHOKED_PEERS_PER_TORRENT; ++i ) {
         tr_peerMsgsSetChoke( data[i].peer->msgs, FALSE );
         ++unchoked;
@@ -1005 +995 @@
     size -= i;
 
-    /* of those remaining, unchoke those that are faster than the downloaders */
-    qsort( data, size, sizeof(ChokeData), compareChokeByRate );
-    for( i=0; i<size && data[i].rate >= bestDownloaderRate; ++i )
-        tr_peerMsgsSetChoke( data[i].peer->msgs, FALSE );
-    memmove( data, data+i, sizeof(ChokeData)*(size-i) );
-    size -= i;
-
-    /* of those remaining, optimistically unchoke one; choke the rest */
-    if( size > 0 ) {
-        const int optimistic = tr_rand( size );
-        for( i=0; i<size; ++i )
-            tr_peerMsgsSetChoke( data[i].peer->msgs, i!=optimistic );
+    /* of those remaining, optimistically the first interested one
+       (the were randomized in qsort already) and choke the rest */
+    for( i=0; i<size; ++i ) {
+        if( optimistic || !data[i].peer->peerIsInterested )
+            tr_peerMsgsSetChoke( data[i].peer->msgs, TRUE );
+        else {
+            optimistic = TRUE;
+            tr_peerMsgsSetChoke( data[i].peer->msgs, FALSE );
+        }
     }
 

branches/encryption/libtransmission/peer-msgs.c

@@ -46 +46 @@
 #define PEER_PULSE_INTERVAL_MSEC 50
 
-/* the most requests we'll batch up for this peer */
-#define OUT_REQUESTS_MAX 5
-
-/* when we get down to this many requests, we ask the manager for more */
-#define OUT_REQUESTS_LOW 3
-
 enum
 {
@@ -81 +75 @@
     uint32_t offset;
     uint32_t length;
+    time_t time_requested;
 };
 
@@ -154 +149 @@
 {
     tr_peermsgs_event e = blankEvent;
-    e.eventType = TR_PEERMSG_GOT_BITFIELD;
+    e.eventType = TR_PEERMSG_PEER_BITFIELD;
     e.bitfield = bitfield;
     tr_publisherPublish( peer->publisher, peer, &e );
@@ -160 +155 @@
 
 static void
-fireGotHave( tr_peermsgs * peer, uint32_t pieceIndex )
+fireHave( tr_peermsgs * msgs, int isClient, uint32_t pieceIndex )
 {
     tr_peermsgs_event e = blankEvent;
-    e.eventType = TR_PEERMSG_GOT_HAVE;
+    e.eventType = isClient ? TR_PEERMSG_CLIENT_HAVE : TR_PEERMSG_PEER_HAVE;
     e.pieceIndex = pieceIndex;
-    tr_publisherPublish( peer->publisher, peer, &e );
+    tr_publisherPublish( msgs->publisher, msgs, &e );
+}
+
+static void
+fireClientHave( tr_peermsgs * msgs, uint32_t pieceIndex )
+{
+    fireHave( msgs, TRUE, pieceIndex );
+}
+
+static void
+firePeerHave( tr_peermsgs * msgs, uint32_t pieceIndex )
+{
+    fireHave( msgs, FALSE, pieceIndex );
 }
 
@@ -172 +179 @@
 {
     tr_peermsgs_event e = blankEvent;
-    e.eventType = TR_PEERMSG_GOT_BLOCK;
+    e.eventType = TR_PEERMSG_CLIENT_BLOCK;
     e.pieceIndex = pieceIndex;
     e.offset = offset;
@@ -236 +243 @@
     const uint8_t bt_msgid = weAreInterested ? BT_INTERESTED : BT_NOT_INTERESTED;
 
-    fprintf( stderr, "peer %p: enqueueing an %s message\n", peer, (weAreInterested ? "interested" : "not interested") );
+    fprintf( stderr, "peer %p: sending an %s message\n", peer, (weAreInterested ? "INTERESTED" : "NOT_INTERESTED") );
     tr_peerIoWriteUint32( peer->io, peer->outMessages, len );
     tr_peerIoWriteBytes( peer->io, peer->outMessages, &bt_msgid, 1 );
@@ -332 +339 @@
 
 int
-tr_peerMsgsAddRequest( tr_peermsgs * peer,
+tr_peerMsgsAddRequest( tr_peermsgs * msgs,
                        uint32_t      index, 
                        uint32_t      offset, 
@@ -340 +347 @@
     const uint32_t len = sizeof(uint8_t) + 3 * sizeof(uint32_t);
     struct peer_request * req;
-
-    if( peer->info->clientIsChoked )
+    int maxSize;
+
+    if( msgs->info->clientIsChoked )
         return TR_ADDREQ_CLIENT_CHOKED;
 
-    if( !tr_bitfieldHas( peer->info->have, index ) )
+    if( !tr_bitfieldHas( msgs->info->have, index ) )
         return TR_ADDREQ_MISSING;
 
-    if( tr_list_size( peer->clientAskedFor) >= OUT_REQUESTS_MAX )
+    maxSize = 2 + (int)(tr_peerIoGetRateToClient(msgs->io)/10);
+    if( maxSize > 100 )
+        maxSize = 100;
+    fprintf( stderr, "peer %p has a max request queue size of %d\n", msgs, maxSize );
+    if( tr_list_size( msgs->clientAskedFor) >= maxSize )
         return TR_ADDREQ_FULL;
 
     /* queue the request */
-    tr_peerIoWriteUint32( peer->io, peer->outMessages, len );
-    tr_peerIoWriteBytes( peer->io, peer->outMessages, &bt_msgid, 1 );
-    tr_peerIoWriteUint32( peer->io, peer->outMessages, index );
-    tr_peerIoWriteUint32( peer->io, peer->outMessages, offset );
-    tr_peerIoWriteUint32( peer->io, peer->outMessages, length );
+    tr_peerIoWriteUint32( msgs->io, msgs->outMessages, len );
+    tr_peerIoWriteBytes( msgs->io, msgs->outMessages, &bt_msgid, 1 );
+    tr_peerIoWriteUint32( msgs->io, msgs->outMessages, index );
+    tr_peerIoWriteUint32( msgs->io, msgs->outMessages, offset );
+    tr_peerIoWriteUint32( msgs->io, msgs->outMessages, length );
     fprintf( stderr, "w00t peer %p: requesting a block from piece %u, offset %u, length %u\n",
-             peer, (unsigned int)index, (unsigned int)offset, (unsigned int)length );
+             msgs, (unsigned int)index, (unsigned int)offset, (unsigned int)length );
 
     /* add it to our `requests sent' list */
@@ -364 +376 @@
     req->offset = offset;
     req->length = length;
-    tr_list_append( &peer->clientAskedFor, req );
-    fprintf( stderr, "w00t added a request; peer %p's clientAskedFor.size() is now %d\n", peer, tr_list_size(peer->clientAskedFor));
+    req->time_requested = time( NULL );
+    tr_list_append( &msgs->clientAskedFor, req );
+    fprintf( stderr, "w00t added a request; peer %p's clientAskedFor.size() is now %d\n", msgs, tr_list_size(msgs->clientAskedFor));
 
     return TR_ADDREQ_OK;
@@ -504 +517 @@
 
 static int
-readBtMessage( tr_peermsgs * peer, struct evbuffer * inbuf )
+readBtMessage( tr_peermsgs * msgs, struct evbuffer * inbuf )
 {
     uint8_t id;
     uint32_t ui32;
-    uint32_t msglen = peer->incomingMessageLength;
+    uint32_t msglen = msgs->incomingMessageLength;
 
     if( EVBUFFER_LENGTH(inbuf) < msglen )
         return READ_MORE;
 
-    tr_peerIoReadBytes( peer->io, inbuf, &id, 1 );
+    tr_peerIoReadBytes( msgs->io, inbuf, &id, 1 );
     msglen--;
     fprintf( stderr, "got a message from the peer... "
@@ -522 +535 @@
         case BT_CHOKE:
             assert( msglen == 0 );
-            fprintf( stderr, "got a BT_CHOKE\n" );
-            peer->info->clientIsChoked = 1;
-            tr_list_foreach( peer->peerAskedFor, tr_free );
-            tr_list_free( &peer->peerAskedFor );
+            fprintf( stderr, "peer-msgs %p sent us a BT_CHOKE\n", msgs );
+            msgs->info->clientIsChoked = 1;
+            tr_list_foreach( msgs->peerAskedFor, tr_free );
+            tr_list_free( &msgs->peerAskedFor );
             /* FIXME: maybe choke them */
             /* FIXME: unmark anything we'd requested from them... */
@@ -532 +545 @@
         case BT_UNCHOKE:
             assert( msglen == 0 );
-            fprintf( stderr, "got a BT_UNCHOKE\n" );
-            peer->info->clientIsChoked = 0;
+            fprintf( stderr, "peer-msgs %p sent us a BT_UNCHOKE\n", msgs );
+            msgs->info->clientIsChoked = 0;
             /* FIXME: maybe unchoke them */
             /* FIXME: maybe send them requests */
@@ -540 +553 @@
         case BT_INTERESTED:
             assert( msglen == 0 );
-            fprintf( stderr, "got a BT_INTERESTED\n" );
-            peer->info->peerIsInterested = 1;
+            fprintf( stderr, "peer-msgs %p sent us a BT_INTERESTED\n", msgs );
+            msgs->info->peerIsInterested = 1;
             /* FIXME: maybe unchoke them */
             break;
@@ -547 +560 @@
         case BT_NOT_INTERESTED:
             assert( msglen == 0 );
-            fprintf( stderr, "got a BT_NOT_INTERESTED\n" );
-            peer->info->peerIsInterested = 0;
+            fprintf( stderr, "peer-msgs %p sent us a BT_NOT_INTERESTED\n", msgs );
+            msgs->info->peerIsInterested = 0;
             /* FIXME: maybe choke them */
             break;
@@ -554 +567 @@
         case BT_HAVE:
             assert( msglen == 4 );
-            fprintf( stderr, "got a BT_HAVE\n" );
-            tr_peerIoReadUint32( peer->io, inbuf, &ui32 );
-            tr_bitfieldAdd( peer->info->have, ui32 );
-            peer->info->progress = tr_bitfieldCountTrueBits( peer->info->have ) / (float)peer->torrent->info.pieceCount;
-            fprintf( stderr, "after the HAVE message, peer progress is %f\n", peer->info->progress );
-            updateInterest( peer );
-            fireGotHave( peer, ui32 );
+            fprintf( stderr, "peer-msgs %p sent us a BT_HAVE\n", msgs );
+            tr_peerIoReadUint32( msgs->io, inbuf, &ui32 );
+            tr_bitfieldAdd( msgs->info->have, ui32 );
+            msgs->info->progress = tr_bitfieldCountTrueBits( msgs->info->have ) / (float)msgs->torrent->info.pieceCount;
+            fprintf( stderr, "after the HAVE message, peer progress is %f\n", msgs->info->progress );
+            updateInterest( msgs );
+            firePeerHave( msgs, ui32 );
             break;
 
         case BT_BITFIELD:
-            assert( msglen == peer->info->have->len );
-            fprintf( stderr, "got a BT_BITFIELD\n" );
-            tr_peerIoReadBytes( peer->io, inbuf, peer->info->have->bits, msglen );
-            peer->info->progress = tr_bitfieldCountTrueBits( peer->info->have ) / (float)peer->torrent->info.pieceCount;
-            fprintf( stderr, "after the BITFIELD peer progress is %f\n", peer->info->progress );
-            fireGotBitfield( peer, peer->info->have );
-            updateInterest( peer );
+            assert( msglen == msgs->info->have->len );
+            fprintf( stderr, "peer-msgs %p sent us a BT_BITFIELD\n", msgs );
+            tr_peerIoReadBytes( msgs->io, inbuf, msgs->info->have->bits, msglen );
+            msgs->info->progress = tr_bitfieldCountTrueBits( msgs->info->have ) / (float)msgs->torrent->info.pieceCount;
+            fprintf( stderr, "after the BITFIELD peer progress is %f\n", msgs->info->progress );
+            fireGotBitfield( msgs, msgs->info->have );
+            updateInterest( msgs );
             /* FIXME: maybe unchoke */
             break;
@@ -579 +592 @@
             fprintf( stderr, "got a BT_REQUEST\n" );
             req = tr_new( struct peer_request, 1 );
-            tr_peerIoReadUint32( peer->io, inbuf, &req->index );
-            tr_peerIoReadUint32( peer->io, inbuf, &req->offset );
-            tr_peerIoReadUint32( peer->io, inbuf, &req->length );
-            if( !peer->info->peerIsChoked )
-                tr_list_prepend( &peer->peerAskedFor, req );
+            tr_peerIoReadUint32( msgs->io, inbuf, &req->index );
+            tr_peerIoReadUint32( msgs->io, inbuf, &req->offset );
+            tr_peerIoReadUint32( msgs->io, inbuf, &req->length );
+            if( !msgs->info->peerIsChoked )
+                tr_list_prepend( &msgs->peerAskedFor, req );
             break;
         }
@@ -592 +605 @@
             assert( msglen == 12 );
             fprintf( stderr, "got a BT_CANCEL\n" );
-            tr_peerIoReadUint32( peer->io, inbuf, &req.index );
-            tr_peerIoReadUint32( peer->io, inbuf, &req.offset );
-            tr_peerIoReadUint32( peer->io, inbuf, &req.length );
-            node = tr_list_find( peer->peerAskedFor, &req, peer_request_compare );
+            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 ) {
                 fprintf( stderr, "found the req that peer is cancelling... cancelled.\n" );
-                tr_list_remove_data( &peer->peerAskedFor, node->data );
+                tr_list_remove_data( &msgs->peerAskedFor, node->data );
             }
             break;
@@ -604 +617 @@
 
         case BT_PIECE: {
-            fprintf( stderr, "got a BT_PIECE\n" );
-            assert( peer->blockToUs.length == 0 );
-            peer->state = READING_BT_PIECE;
-            tr_peerIoReadUint32( peer->io, inbuf, &peer->blockToUs.index );
-            tr_peerIoReadUint32( peer->io, inbuf, &peer->blockToUs.offset );
-            peer->blockToUs.length = msglen - 8;
-fprintf( stderr, "left to read is [%"PRIu64"]\n", (uint64_t)peer->blockToUs.length );
-            assert( peer->blockToUs.length > 0 );
-            evbuffer_drain( peer->inBlock, ~0 );
+            fprintf( stderr, "peer-msgs %p sent us a BT_PIECE\n", msgs );
+            assert( msgs->blockToUs.length == 0 );
+            msgs->state = READING_BT_PIECE;
+            tr_peerIoReadUint32( msgs->io, inbuf, &msgs->blockToUs.index );
+            tr_peerIoReadUint32( msgs->io, inbuf, &msgs->blockToUs.offset );
+            msgs->blockToUs.length = msglen - 8;
+            assert( msgs->blockToUs.length > 0 );
+            evbuffer_drain( msgs->inBlock, ~0 );
             return READ_AGAIN;
             break;
@@ -619 +631 @@
         case BT_PORT: {
             assert( msglen == 2 );
-            fprintf( stderr, "got a BT_PORT\n" );
-            tr_peerIoReadUint16( peer->io, inbuf, &peer->listeningPort );
+            fprintf( stderr, "peer-msgs %p sent us a BT_PORT\n", msgs );
+            tr_peerIoReadUint16( msgs->io, inbuf, &msgs->listeningPort );
             break;
         }
 
         case BT_LTEP:
-            fprintf( stderr, "got a BT_LTEP\n" );
-            parseLtep( peer, msglen, inbuf );
+            fprintf( stderr, "peer-msgs %p sent us a BT_LTEP\n", msgs );
+            parseLtep( msgs, msglen, inbuf );
             break;
 
         default:
-            fprintf( stderr, "got an unknown BT message type: %d\n", (int)id );
-            tr_peerIoDrain( peer->io, inbuf, msglen );
+            fprintf( stderr, "peer-msgs %p sent us an UNKNOWN: %d\n", msgs, (int)id );
+            tr_peerIoDrain( msgs->io, inbuf, msglen );
             assert( 0 );
     }
 
-    peer->incomingMessageLength = -1;
-    peer->state = AWAITING_BT_LENGTH;
+    msgs->incomingMessageLength = -1;
+    msgs->state = AWAITING_BT_LENGTH;
     return READ_AGAIN;
 }
 
+static void
+clientGotBytes( tr_peermsgs * msgs, uint32_t byteCount )
+{
+    tr_torrent * tor = msgs->torrent;
+    tor->downloadedCur += byteCount;
+    tr_rcTransferred( tor->download, byteCount );
+    tr_rcTransferred( tor->handle->download, byteCount );
+}
+
+static void
+peerGotBytes( tr_peermsgs * msgs, uint32_t byteCount )
+{
+    tr_torrent * tor = msgs->torrent;
+    tor->uploadedCur += byteCount;
+    tr_rcTransferred( tor->upload, byteCount );
+    tr_rcTransferred( tor->upload, byteCount );
+}
+
 static int
-canDownload( const tr_peermsgs * peer UNUSED )
+canDownload( const tr_peermsgs * msgs UNUSED )
 {
 #if 0
-    tr_torrent * tor = peer->torrent;
+    tr_torrent * tor = msgs->torrent;
 
     if( tor->downloadLimitMode == TR_SPEEDLIMIT_GLOBAL )
@@ -657 +687 @@
 
 static void
-gotBlock( tr_peermsgs * peer, int index, int offset, struct evbuffer * inbuf )
-{
-    tr_torrent * tor = peer->torrent;
-    const size_t length = EVBUFFER_LENGTH( inbuf );
+reassignBytesToCorrupt( tr_peermsgs * msgs, uint32_t byteCount )
+{
+    tr_torrent * tor = msgs->torrent;
+
+    /* increment the `corrupt' field */
+    tor->corruptCur += byteCount;
+
+    /* decrement the `downloaded' field */
+    if( tor->downloadedCur >= byteCount )
+        tor->downloadedCur -= byteCount;
+    else
+        tor->downloadedCur = 0;
+}
+
+
+static void
+gotBadPiece( tr_peermsgs * msgs, uint32_t pieceIndex )
+{
+    const uint32_t byteCount = tr_torPieceCountBytes( msgs->torrent, (int)pieceIndex );
+    reassignBytesToCorrupt( msgs, byteCount );
+}
+
+static void
+gotUnwantedBlock( tr_peermsgs * msgs, uint32_t index UNUSED, uint32_t offset UNUSED, uint32_t length )
+{
+    reassignBytesToCorrupt( msgs, length );
+}
+
+static void
+addUsToBlamefield( tr_peermsgs * msgs, uint32_t index )
+{
+    if( !msgs->info->blame )
+         msgs->info->blame = tr_bitfieldNew( msgs->torrent->info.pieceCount );
+    tr_bitfieldAdd( msgs->info->blame, index );
+}
+
+static void
+gotBlock( tr_peermsgs      * msgs,
+          struct evbuffer  * inbuf,
+          uint32_t           index,
+          uint32_t           offset,
+          uint32_t           length )
+{
+    tr_torrent * tor = msgs->torrent;
     const int block = _tr_block( tor, index, offset );
     struct peer_request key, *req;
 
-    /* remove it from our `we asked for this' list */
+    /**
+    *** Remove the block from our `we asked for this' list
+    **/
+
     key.index = index;
     key.offset = offset;
     key.length = length;
-    req = (struct peer_request*) tr_list_remove( &peer->clientAskedFor, &key,
+    req = (struct peer_request*) tr_list_remove( &msgs->clientAskedFor, &key,
                                                  peer_request_compare );
+    fprintf( stderr, "w00t got a block from %p. turnaround time for this block was %d seconds\n",
+                     msgs, (int)(time(NULL) - req->time_requested) );
     if( req == NULL ) {
-fprintf( stderr, "we didn't ask for this message...\n" );
+        gotUnwantedBlock( msgs, index, offset, length );
+        fprintf( stderr, "we didn't ask for this message...\n" );
         tr_dbg( "we didn't ask the peer for this message..." );
         return;
     }
     tr_free( req );
-    fprintf( stderr, "peer %p now has %d block requests in its outbox\n", peer, tr_list_size(peer->clientAskedFor));
-
-    /* sanity clause */
+    fprintf( stderr, "peer %p now has %d block requests in its outbox\n",
+             msgs, tr_list_size(msgs->clientAskedFor));
+
+    /**
+    *** Error checks
+    **/
+
     if( tr_cpBlockIsComplete( tor->completion, block ) ) {
-fprintf( stderr, "have this block already...\n" );
+        fprintf( stderr, "have this block already...\n" );
         tr_dbg( "have this block already..." );
+        gotUnwantedBlock( msgs, index, offset, length );
         return;
     }
+
     if( (int)length != tr_torBlockCountBytes( tor, block ) ) {
-fprintf( stderr, "block is the wrong length... expected %d and got %d\n", (int)length, (int)tr_torBlockCountBytes(tor,block) );
+        fprintf( stderr, "block is the wrong length..." );
         tr_dbg( "block is the wrong length..." );
+        gotUnwantedBlock( msgs, index, offset, length );
         return;
     }
 
-    fireGotBlock( peer, index, offset, length );
-
-    /* write to disk */
-    if( tr_ioWrite( tor, index, offset, length, EVBUFFER_DATA( inbuf )))
+    /**
+    ***  Write the block
+    **/
+
+    if( tr_ioWrite( tor, index, offset, length, EVBUFFER_DATA( inbuf ))) {
         return;
-
-    /* make a note that this peer helped us with this piece */
-    if( !peer->info->blame )
-         peer->info->blame = tr_bitfieldNew( tor->info.pieceCount );
-    tr_bitfieldAdd( peer->info->blame, index );
+    }
 
     tr_cpBlockAdd( tor->completion, block );
 
-    tor->downloadedCur += length;
-    tr_rcTransferred( tor->download, length );
-    tr_rcTransferred( tor->handle->download, length );
+    addUsToBlamefield( msgs, index );
+
+    fireGotBlock( msgs, index, offset, length );
+
+    fireBlocksRunningLow( msgs );
+
+    /**
+    ***  Handle if this was the last block in the piece
+    **/
+
+    if( tr_cpPieceIsComplete( tor->completion, index ) )
+    {
+        if( !tr_ioHash( tor, index ) )
+        {
+            gotBadPiece( msgs, index );
+            return;
+        }
+
+        fireClientHave( msgs, index );
+    }
 }
 
 
 static ReadState
-readBtPiece( tr_peermsgs * peer, struct evbuffer * inbuf )
-{
-    assert( peer->blockToUs.length > 0 );
-
-    if( !canDownload( peer ) )
-    {
-        peer->notListening = 1;
-        tr_peerIoSetIOMode ( peer->io, 0, EV_READ );
+readBtPiece( tr_peermsgs * msgs, struct evbuffer * inbuf )
+{
+    assert( msgs->blockToUs.length > 0 );
+
+    if( !canDownload( msgs ) )
+    {
+        msgs->notListening = 1;
+        tr_peerIoSetIOMode ( msgs->io, 0, EV_READ );
         return READ_DONE;
     }
@@ -723 +820 @@
     {
         /* inbuf ->  inBlock */
-        const uint32_t len = MIN( EVBUFFER_LENGTH(inbuf), peer->blockToUs.length );
+        const uint32_t len = MIN( EVBUFFER_LENGTH(inbuf), msgs->blockToUs.length );
         uint8_t * tmp = tr_new( uint8_t, len );
-        tr_peerIoReadBytes( peer->io, inbuf, tmp, len );
-        evbuffer_add( peer->inBlock, tmp, len );
+        tr_peerIoReadBytes( msgs->io, inbuf, tmp, len );
+        evbuffer_add( msgs->inBlock, tmp, len );
         tr_free( tmp );
-        peer->blockToUs.length -= len;
-fprintf( stderr, "got %"PRIu64"; left to read is [%"PRIu64"]\n", (uint64_t)len, (uint64_t)peer->blockToUs.length );
-
-
-        if( !peer->blockToUs.length )
+        msgs->blockToUs.length -= len;
+        clientGotBytes( msgs, len );
+fprintf( stderr, "got %"PRIu64"; left to read is [%"PRIu64"]\n", (uint64_t)len, (uint64_t)msgs->blockToUs.length );
+
+        if( !msgs->blockToUs.length )
         {
-fprintf( stderr, "w00t -- index %u, offset %u\n", peer->blockToUs.index, peer->blockToUs.offset );
-            gotBlock( peer, peer->blockToUs.index,
-                            peer->blockToUs.offset,
-                            peer->inBlock );
-            evbuffer_drain( peer->outBlock, ~0 );
-            peer->state = AWAITING_BT_LENGTH;
+fprintf( stderr, "w00t -- index %u, offset %u\n", msgs->blockToUs.index, msgs->blockToUs.offset );
+            gotBlock( msgs, msgs->inBlock,
+                            msgs->blockToUs.index,
+                            msgs->blockToUs.offset,
+                            EVBUFFER_LENGTH( msgs->inBlock ) );
+            evbuffer_drain( msgs->inBlock, ~0 );
+            msgs->state = AWAITING_BT_LENGTH;
         }
 
@@ -795 +893 @@
         tr_peerIoSetIOMode ( msgs->io, EV_READ, 0 );
     }
-
-    /* if we're running low on requests, ask for ones */
-    if( tr_list_size(msgs->clientAskedFor) <= OUT_REQUESTS_LOW )
-        fireBlocksRunningLow( msgs );
 
     if(( len = EVBUFFER_LENGTH( msgs->outBlock ) ))
@@ -807 +901 @@
             tr_peerIoWrite( msgs->io, EVBUFFER_DATA(msgs->outBlock), outlen );
             evbuffer_drain( msgs->outBlock, outlen );
-
-            msgs->torrent->uploadedCur += outlen;
-            tr_rcTransferred( msgs->torrent->upload, outlen );
-            tr_rcTransferred( msgs->handle->upload, outlen );
-
+            peerGotBytes( msgs, outlen );
             len -= outlen;
         }
@@ -834 +924 @@
         tr_free( tmp );
     }
+
+    fireBlocksRunningLow( msgs );
 
     return TRUE; /* loop forever */
@@ -986 +1078 @@
         /* write the pex message */
         benc = tr_bencSaveMalloc( &val, &bencLen );
-        tr_peerIoWriteUint32( peer->io, peer->outBlock, 1 + 1 + bencLen );
-        tr_peerIoWriteBytes ( peer->io, peer->outBlock, &bt_msgid, 1 );
-        tr_peerIoWriteBytes ( peer->io, peer->outBlock, &ltep_msgid, 1 );
-        tr_peerIoWriteBytes ( peer->io, peer->outBlock, benc, bencLen );
+        tr_peerIoWriteUint32( peer->io, peer->outMessages, 1 + 1 + bencLen );
+        tr_peerIoWriteBytes ( peer->io, peer->outMessages, &bt_msgid, 1 );
+        tr_peerIoWriteBytes ( peer->io, peer->outMessages, &ltep_msgid, 1 );
+        tr_peerIoWriteBytes ( peer->io, peer->outMessages, benc, bencLen );
 
         /* cleanup */

branches/encryption/libtransmission/peer-msgs.h

@@ -57 +57 @@
 typedef enum
 {
-    TR_PEERMSG_GOT_BITFIELD,
-    TR_PEERMSG_GOT_HAVE,
-    TR_PEERMSG_GOT_BLOCK,
+    TR_PEERMSG_PEER_BITFIELD,
+    TR_PEERMSG_PEER_HAVE,
+    TR_PEERMSG_CLIENT_HAVE,
+    TR_PEERMSG_CLIENT_BLOCK,
     TR_PEERMSG_GOT_PEX,
     TR_PEERMSG_GOT_ERROR,

branches/encryption/libtransmission/shared.c

@@ -82 +82 @@
     s->natpmp     = tr_natpmpInit();
     s->upnp       = tr_upnpInit();
-    s->pulseTimer   = tr_timerNew( h, SharedLoop, s, 200 );
+    s->pulseTimer   = tr_timerNew( h, SharedLoop, s, 250 );
 
     return s;