Changeset 7176 for branches/1.4x/libtransmission/peer-msgs.c

Timestamp:

Nov 29, 2008, 4:44:24 PM (13 years ago)

Author:

charles

Message:

(1.4x libT) backport the bandwidth fixes back to the 1.4x branch, pass #2: most of the bandwidth code, plus autoconf and xcode makefile stuff

File:

• branches/1.4x/libtransmission/peer-msgs.c (modified) (33 diffs)

branches/1.4x/libtransmission/peer-msgs.c

@@ -25 +25 @@
 #include "crypto.h"
 #include "inout.h"
+#include "iobuf.h"
 #ifdef WIN32
 #include "net.h" /* for ECONN */
@@ -261 +262 @@
 struct tr_peermsgs
 {
-    unsigned int    peerSentBitfield        : 1;
-    unsigned int    peerSupportsPex         : 1;
-    unsigned int    clientSentLtepHandshake : 1;
-    unsigned int    peerSentLtepHandshake   : 1;
-    unsigned int    sendingBlock            : 1;
+    tr_bool         peerSentBitfield;
+    tr_bool         peerSupportsPex;
+    tr_bool         clientSentLtepHandshake;
+    tr_bool         peerSentLtepHandshake;
 
     uint8_t         state;
@@ -286 +286 @@
     tr_publisher_t *       publisher;
 
-    struct evbuffer *      outBlock; /* buffer of the current piece message */
     struct evbuffer *      outMessages; /* all the non-piece messages */
 
@@ -345 +344 @@
 }
 
-#define dbgmsg( msgs, ... ) myDebug( __FILE__, __LINE__, msgs, __VA_ARGS__ )
+#define dbgmsg( msgs, ... ) \
+    do { \
+        if( tr_deepLoggingIsActive( ) ) \
+            myDebug( __FILE__, __LINE__, msgs, __VA_ARGS__ ); \
+    } while( 0 )
 
 /**
@@ -430 +433 @@
 **/
 
-static const tr_peer_event blankEvent = { 0, 0, 0, 0, 0.0f, 0 };
+static const tr_peer_event blankEvent = { 0, 0, 0, 0, 0.0f, 0, 0 };
 
 static void
@@ -484 +487 @@
 static void
 fireClientGotData( tr_peermsgs * msgs,
-                   uint32_t      length )
+                   uint32_t      length,
+                   int           wasPieceData )
 {
     tr_peer_event e = blankEvent;
@@ -490 +494 @@
     e.length = length;
     e.eventType = TR_PEER_CLIENT_GOT_DATA;
+    e.wasPieceData = wasPieceData;
     publish( msgs, &e );
 }
 
 static void
-firePeerGotData( tr_peermsgs * msgs,
-                 uint32_t      length )
+firePeerGotData( tr_peermsgs  * msgs,
+                 uint32_t       length,
+                 int            wasPieceData )
 {
     tr_peer_event e = blankEvent;
@@ -501 +507 @@
     e.length = length;
     e.eventType = TR_PEER_PEER_GOT_DATA;
+    e.wasPieceData = wasPieceData;
+
     publish( msgs, &e );
 }
@@ -906 +914 @@
     **/
 
-    dbgmsg( msgs, "added req for piece %lu", (unsigned long)index );
+    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 );
@@ -918 +927 @@
     struct request_list a = msgs->clientWillAskFor;
     struct request_list b = msgs->clientAskedFor;
+    dbgmsg( msgs, "cancelling all requests to peer" );
 
     msgs->clientAskedFor = REQUEST_LIST_INIT;
@@ -945 +955 @@
     assert( length > 0 );
 
+
     /* have we asked the peer for this piece? */
     req.index = pieceIndex;
@@ -951 +962 @@
 
     /* if it's only in the queue and hasn't been sent yet, free it */
-    if( reqListRemove( &msgs->clientWillAskFor, &req ) )
+    if( reqListRemove( &msgs->clientWillAskFor, &req ) ) {
+        dbgmsg( msgs, "cancelling %"PRIu32":%"PRIu32"->%"PRIu32"\n", 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"\n", pieceIndex, offset, length );
         protocolSendCancel( msgs, &req );
         fireCancelledReq( msgs, &req );
@@ -1029 +1043 @@
     }
 
-    dbgmsg( msgs, "here is the ltep handshake we got [%*.*s]", len, len,
-            tmp );
+    dbgmsg( msgs, "here is the handshake: [%*.*s]", len, len,  tmp );
 
     /* does the peer prefer encrypted connections? */
@@ -1300 +1313 @@
                            const struct peer_request * req );
 
-static void
-clientGotBytes( tr_peermsgs * msgs,
-                uint32_t      byteCount )
-{
-    msgs->info->pieceDataActivityDate = time( NULL );
-    fireClientGotData( msgs, byteCount );
-}
-
 static int
-readBtPiece( tr_peermsgs *     msgs,
-             struct evbuffer * inbuf,
-             size_t            inlen )
+readBtPiece( tr_peermsgs      * msgs,
+             struct evbuffer  * inbuf,
+             size_t             inlen,
+             size_t           * setme_piece_bytes_read )
 {
     struct peer_request * req = &msgs->incoming.blockReq;
@@ -1343 +1349 @@
         tr_peerIoReadBytes( msgs->io, inbuf, buf, n );
         evbuffer_add( msgs->incoming.block, buf, n );
-        clientGotBytes( msgs, n );
+        fireClientGotData( msgs, n, TRUE );
+        *setme_piece_bytes_read += n;
         tr_free( buf );
         dbgmsg( msgs, "got %d bytes for block %u:%u->%u ... %d remain",
@@ -1540 +1547 @@
 
 static void
-peerGotBytes( tr_peermsgs * msgs,
-              uint32_t      byteCount,
-              const time_t  now )
-{
-    msgs->info->pieceDataActivityDate = now;
-    firePeerGotData( msgs, byteCount );
-}
-
-static void
 decrementDownloadedCount( tr_peermsgs * msgs,
                           uint32_t      byteCount )
@@ -1635 +1633 @@
 }
 
+static int peerPulse( void * vmsgs );
+
+static void
+didWrite( tr_peerIo * io UNUSED, size_t bytesWritten, int wasPieceData, void * vmsgs )
+{
+    tr_peermsgs * msgs = vmsgs;
+    firePeerGotData( msgs, bytesWritten, wasPieceData );
+    peerPulse( msgs );
+}
+
 static ReadState
-canRead( struct bufferevent * evin,
-         void *               vmsgs )
+canRead( struct tr_iobuf * iobuf, void * vmsgs, size_t * piece )
 {
     ReadState         ret;
     tr_peermsgs *     msgs = vmsgs;
-    struct evbuffer * in = EVBUFFER_INPUT ( evin );
+    struct evbuffer * in = tr_iobuf_input( iobuf );
     const size_t      inlen = EVBUFFER_LENGTH( in );
 
@@ -1650 +1657 @@
     else if( msgs->state == AWAITING_BT_PIECE )
     {
-        ret = inlen ? readBtPiece( msgs, in, inlen ) : READ_LATER;
+        ret = inlen ? readBtPiece( msgs, in, inlen, piece ) : READ_LATER;
     }
     else switch( msgs->state )
-        {
-            case AWAITING_BT_LENGTH:
-                ret = readBtLength ( msgs, in, inlen ); break;
-
-            case AWAITING_BT_ID:
-                ret = readBtId     ( msgs, in, inlen ); break;
-
-            case AWAITING_BT_MESSAGE:
-                ret = readBtMessage( msgs, in, inlen ); break;
-
-            default:
-                assert( 0 );
-        }
+    {
+        case AWAITING_BT_LENGTH:
+            ret = readBtLength ( msgs, in, inlen ); break;
+
+        case AWAITING_BT_ID:
+            ret = readBtId     ( msgs, in, inlen ); break;
+
+        case AWAITING_BT_MESSAGE:
+            ret = readBtMessage( msgs, in, inlen ); break;
+
+        default:
+            assert( 0 );
+    }
+
+    /* log the raw data that was read */
+    if( EVBUFFER_LENGTH( in ) != inlen )
+        fireClientGotData( msgs, inlen - EVBUFFER_LENGTH( in ), FALSE );
 
     return ret;
-}
-
-static void
-sendKeepalive( tr_peermsgs * msgs )
-{
-    dbgmsg( msgs, "sending a keepalive message" );
-    tr_peerIoWriteUint32( msgs->io, msgs->outMessages, 0 );
-    pokeBatchPeriod( msgs, IMMEDIATE_PRIORITY_INTERVAL_SECS );
 }
 
@@ -1691 +1694 @@
                   ( rateToClient * 30 * 1024 ) / peer->torrent->blockSize;
 
-    peer->minActiveRequests = 4;
-    peer->maxActiveRequests = peer->minActiveRequests +
-                              estimatedBlocksInNext30Seconds;
+    peer->minActiveRequests = 8;
+    peer->maxActiveRequests = peer->minActiveRequests + estimatedBlocksInNext30Seconds;
     return TRUE;
 }
@@ -1705 +1707 @@
 }
 
+static size_t
+fillOutputBuffer( tr_peermsgs * msgs, time_t now )
+{
+    size_t bytesWritten = 0;
+    struct peer_request req;
+    const int haveMessages = EVBUFFER_LENGTH( msgs->outMessages ) != 0;
+
+    /**
+    ***  Protocol messages
+    **/
+
+    if( haveMessages && !msgs->outMessagesBatchedAt ) /* fresh batch */
+    {
+        dbgmsg( msgs, "started an outMessages batch (length is %zu)", EVBUFFER_LENGTH( msgs->outMessages ) );
+        msgs->outMessagesBatchedAt = now;
+    }
+    else if( haveMessages && ( ( now - msgs->outMessagesBatchedAt ) >= msgs->outMessagesBatchPeriod ) )
+    {
+        const size_t len = EVBUFFER_LENGTH( msgs->outMessages );
+        /* flush the protocol messages */
+        dbgmsg( msgs, "flushing outMessages... to %p (length is %zu)", msgs->io, len );
+        tr_peerIoWriteBuf( msgs->io, msgs->outMessages, FALSE );
+        msgs->clientSentAnythingAt = now;
+        msgs->outMessagesBatchedAt = 0;
+        msgs->outMessagesBatchPeriod = LOW_PRIORITY_INTERVAL_SECS;
+        bytesWritten +=  len;
+    }
+
+    /**
+    ***  Blocks
+    **/
+
+    if( ( tr_peerIoGetWriteBufferSpace( msgs->io ) >= msgs->torrent->blockSize )
+        && popNextRequest( msgs, &req )
+        && requestIsValid( msgs, &req )
+        && tr_cpPieceIsComplete( msgs->torrent->completion, req.index ) )
+    {
+        /* send a block */
+        uint8_t * buf = tr_new( uint8_t, req.length );
+        const int err = tr_ioRead( msgs->torrent, req.index, req.offset, req.length, buf );
+        if( err ) {
+            fireError( msgs, err );
+        } else {
+            tr_peerIo * io = msgs->io;
+            struct evbuffer * out = evbuffer_new( );
+            dbgmsg( msgs, "sending block %u:%u->%u", req.index, req.offset, req.length );
+            tr_peerIoWriteUint32( io, out, sizeof( uint8_t ) + 2 * sizeof( uint32_t ) + req.length );
+            tr_peerIoWriteUint8 ( io, out, BT_PIECE );
+            tr_peerIoWriteUint32( io, out, req.index );
+            tr_peerIoWriteUint32( io, out, req.offset );
+            tr_peerIoWriteBytes ( io, out, buf, req.length );
+            tr_peerIoWriteBuf( io, out, TRUE );
+            bytesWritten += EVBUFFER_LENGTH( out );
+            evbuffer_free( out );
+            msgs->clientSentAnythingAt = now;
+        }
+        tr_free( buf );
+    }
+
+    /**
+    ***  Keepalive
+    **/
+
+    if( msgs->clientSentAnythingAt
+        && ( ( now - msgs->clientSentAnythingAt ) > KEEPALIVE_INTERVAL_SECS ) )
+    {
+        dbgmsg( msgs, "sending a keepalive message" );
+        tr_peerIoWriteUint32( msgs->io, msgs->outMessages, 0 );
+        pokeBatchPeriod( msgs, IMMEDIATE_PRIORITY_INTERVAL_SECS );
+    }
+
+    return bytesWritten;
+}
+
 static int
 peerPulse( void * vmsgs )
@@ -1713 +1789 @@
     ratePulse( msgs );
 
-    /*tr_peerIoTryRead( msgs->io );*/
     pumpRequestQueue( msgs, now );
     expireOldRequests( msgs, now );
 
-    if( msgs->sendingBlock )
-    {
-        const size_t uploadMax = tr_peerIoGetWriteBufferSpace( msgs->io );
-        size_t       len = EVBUFFER_LENGTH( msgs->outBlock );
-        const size_t outlen = MIN( len, uploadMax );
-
-        assert( len );
-
-        if( outlen )
-        {
-            tr_peerIoWrite( msgs->io, EVBUFFER_DATA( msgs->outBlock ), outlen );
-            evbuffer_drain( msgs->outBlock, outlen );
-            peerGotBytes( msgs, outlen, now );
-
-            len -= outlen;
-            msgs->clientSentAnythingAt = now;
-            msgs->sendingBlock = len != 0;
-
-            dbgmsg( msgs, "wrote %d bytes; %d left in block", (int)outlen,
-                    (int)len );
-        }
-        else dbgmsg( msgs,
-                     "stalled writing block... uploadMax %lu, outlen %lu",
-                     uploadMax, outlen );
-    }
-
-    if( !msgs->sendingBlock )
-    {
-        struct peer_request req;
-        const int haveMessages = EVBUFFER_LENGTH( msgs->outMessages ) != 0;
-
-        if( haveMessages && !msgs->outMessagesBatchedAt ) /* fresh batch */
-        {
-            dbgmsg( msgs, "started an outMessages batch (length is %d)",
-                   (int)EVBUFFER_LENGTH( msgs->outMessages ) );
-            msgs->outMessagesBatchedAt = now;
-        }
-        else if( haveMessages
-               && ( ( now - msgs->outMessagesBatchedAt ) >
-                   msgs->outMessagesBatchPeriod ) )
-        {
-            dbgmsg( msgs, "flushing outMessages... (length is %d)",
-                   (int)EVBUFFER_LENGTH(
-                       msgs->outMessages ) );
-            tr_peerIoWriteBuf( msgs->io, msgs->outMessages );
-            msgs->clientSentAnythingAt = now;
-            msgs->outMessagesBatchedAt = 0;
-            msgs->outMessagesBatchPeriod = LOW_PRIORITY_INTERVAL_SECS;
-        }
-        else if( !EVBUFFER_LENGTH( msgs->outBlock )
-               && popNextRequest( msgs, &req )
-               && requestIsValid( msgs, &req )
-               && tr_cpPieceIsComplete( msgs->torrent->completion,
-                                        req.index ) )
-        {
-            uint8_t * buf = tr_new( uint8_t, req.length );
-            const int err = tr_ioRead( msgs->torrent,
-                                       req.index, req.offset, req.length,
-                                       buf );
-            if( err )
-            {
-                fireError( msgs, err );
-            }
-            else
-            {
-                tr_peerIo *       io = msgs->io;
-                struct evbuffer * out = msgs->outBlock;
-
-                dbgmsg( msgs, "sending block %u:%u->%u", req.index,
-                        req.offset,
-                        req.length );
-                tr_peerIoWriteUint32(
-                    io, out, sizeof( uint8_t ) + 2 * sizeof( uint32_t ) +
-                    req.length );
-                tr_peerIoWriteUint8 ( io, out, BT_PIECE );
-                tr_peerIoWriteUint32( io, out, req.index );
-                tr_peerIoWriteUint32( io, out, req.offset );
-                tr_peerIoWriteBytes ( io, out, buf, req.length );
-                msgs->sendingBlock = 1;
-            }
-
-            tr_free( buf );
-        }
-        else if( ( !haveMessages )
-               && ( now - msgs->clientSentAnythingAt ) >
-                KEEPALIVE_INTERVAL_SECS )
-        {
-            sendKeepalive( msgs );
-        }
-    }
+    for( ;; )
+        if( fillOutputBuffer( msgs, now ) < 1 )
+            break;
 
     return TRUE; /* loop forever */
@@ -1819 +1807 @@
 
 static void
-gotError( struct bufferevent * evbuf UNUSED,
-          short                      what,
-          void *                     vmsgs )
+gotError( struct tr_iobuf  * iobuf UNUSED,
+          short              what,
+          void             * vmsgs )
 {
     if( what & EVBUFFER_TIMEOUT )
-        dbgmsg( vmsgs, "libevent got a timeout, what=%hd, secs=%d", what,
-                evbuf->timeout_read );
+        dbgmsg( vmsgs, "libevent got a timeout, what=%hd", what );
     if( what & ( EVBUFFER_EOF | EVBUFFER_ERROR ) )
         dbgmsg( vmsgs, "libevent got an error! what=%hd, errno=%d (%s)",
@@ -1843 +1830 @@
     field = tr_bitfieldDup( tr_cpPieceBitfield( msgs->torrent->completion ) );
 
+#if 0
     if( tr_sessionIsLazyBitfieldEnabled( msgs->session ) )
     {
@@ -1848 +1836 @@
             speed over a truly random sample -- let's limit the pool size to
             the first 1000 pieces so large torrents don't bog things down */
-        size_t             poolSize = MIN( msgs->torrent->info.pieceCount,
-                                           1000 );
-        tr_piece_index_t * pool = tr_new( tr_piece_index_t, poolSize );
+        size_t poolSize;
+        const size_t maxPoolSize = MIN( msgs->torrent->info.pieceCount, 1000 );
+        tr_piece_index_t * pool = tr_new( tr_piece_index_t, maxPoolSize );
 
         /* build the pool */
-        for( i = 0; i < poolSize; ++i )
-            pool[i] = i;
+        for( i=poolSize=0; i<maxPoolSize; ++i )
+            if( tr_bitfieldHas( field, i ) )
+                pool[poolSize++] = i;
 
         /* pull random piece indices from the pool */
         while( ( poolSize > 0 ) && ( lazyCount < LAZY_PIECE_COUNT ) )
         {
-            const int              pos = tr_cryptoWeakRandInt( poolSize );
+            const int pos = tr_cryptoWeakRandInt( poolSize );
             const tr_piece_index_t piece = pool[pos];
             tr_bitfieldRem( field, piece );
@@ -1869 +1858 @@
         tr_free( pool );
     }
+#endif
 
     tr_peerIoWriteUint32( msgs->io, out,
@@ -1947 +1937 @@
     if( msgs->peerSupportsPex && tr_torrentAllowsPex( msgs->torrent ) )
     {
-        PexDiffs          diffs;
-        tr_pex *          newPex = NULL;
-        const int         newCount = tr_peerMgrGetPeers( msgs->session->peerMgr,
-                                                         msgs->torrent->info.hash,
-                                                         &newPex );
+        PexDiffs diffs;
+        tr_pex * newPex = NULL;
+        const int newCount = tr_peerMgrGetPeers( msgs->session->peerMgr,
+                                                 msgs->torrent->info.hash,
+                                                 &newPex );
+
         /* build the diffs */
         diffs.added = tr_new( tr_pex, newCount );
@@ -1970 +1961 @@
         if( diffs.addedCount || diffs.droppedCount )
         {
-            int               i;
-            tr_benc           val;
-            uint8_t *         tmp, *walk;
-            char *            benc;
-            int               bencLen;
+            int  i;
+            tr_benc val;
+            char * benc;
+            int bencLen;
+            uint8_t * tmp, *walk;
             struct evbuffer * out = msgs->outMessages;
 
@@ -1987 +1978 @@
             /* "added" */
             tmp = walk = tr_new( uint8_t, diffs.addedCount * 6 );
-            for( i = 0; i < diffs.addedCount; ++i )
-            {
+            for( i = 0; i < diffs.addedCount; ++i ) {
                 memcpy( walk, &diffs.added[i].in_addr, 4 ); walk += 4;
                 memcpy( walk, &diffs.added[i].port, 2 ); walk += 2;
@@ -2006 +1996 @@
             /* "dropped" */
             tmp = walk = tr_new( uint8_t, diffs.droppedCount * 6 );
-            for( i = 0; i < diffs.droppedCount; ++i )
-            {
+            for( i = 0; i < diffs.droppedCount; ++i ) {
                 memcpy( walk, &diffs.dropped[i].in_addr, 4 ); walk += 4;
                 memcpy( walk, &diffs.dropped[i].port, 2 ); walk += 2;
@@ -2021 +2010 @@
             tr_peerIoWriteUint8 ( msgs->io, out, msgs->ut_pex_id );
             tr_peerIoWriteBytes ( msgs->io, out, benc, bencLen );
-            pokeBatchPeriod( msgs, IMMEDIATE_PRIORITY_INTERVAL_SECS );
-            dbgmsg( msgs, "outMessage size is now %zu", EVBUFFER_LENGTH( out ) );
-
-            /* cleanup */
+            pokeBatchPeriod( msgs, HIGH_PRIORITY_INTERVAL_SECS );
+            dbgmsg( msgs, "sending a pex message; outMessage size is now %zu", EVBUFFER_LENGTH( out ) );
+
             tr_free( benc );
             tr_bencFree( &val );
         }
 
+        /* cleanup */
         tr_free( diffs.added );
         tr_free( diffs.dropped );
@@ -2077 +2066 @@
     m->outMessagesBatchPeriod = LOW_PRIORITY_INTERVAL_SECS;
     m->incoming.block = evbuffer_new( );
-    m->outBlock = evbuffer_new( );
     m->peerAllowedPieces = NULL;
     m->peerAskedFor = REQUEST_LIST_INIT;
@@ -2092 +2080 @@
     tr_peerIoSetTimeoutSecs( m->io, 150 ); /* timeout after N seconds of
                                              inactivity */
-    tr_peerIoSetIOFuncs( m->io, canRead, NULL, gotError, m );
+    tr_peerIoSetIOFuncs( m->io, canRead, didWrite, gotError, m );
     ratePulse( m );
 
@@ -2112 +2100 @@
         evbuffer_free( msgs->incoming.block );
         evbuffer_free( msgs->outMessages );
-        evbuffer_free( msgs->outBlock );
         tr_free( msgs->pex );