Changeset 7588

Timestamp:

Jan 2, 2009, 11:28:57 PM (12 years ago)

Author:

charles

Message:

(trunk libT) revert r7548, which broke very low speed download limits.. the simplified peer-msgs parsing didn't distinguish between piece & raw data until the piece was done downloading.

File:

• trunk/libtransmission/peer-msgs.c (modified) (11 diffs)

trunk/libtransmission/peer-msgs.c

@@ -107 +107 @@
 **/
 
+enum
+{
+    AWAITING_BT_LENGTH,
+    AWAITING_BT_ID,
+    AWAITING_BT_MESSAGE,
+    AWAITING_BT_PIECE
+};
+
 struct peer_request
 {
@@ -245 +253 @@
 struct tr_incoming
 {
-    uint32_t          length; /* includes the +1 for id length */
-    struct evbuffer * block; /* piece data for incoming blocks */
+    uint8_t                id;
+    uint32_t               length; /* includes the +1 for id length */
+    struct peer_request    blockReq; /* metadata for incoming blocks */
+    struct evbuffer *      block; /* piece data for incoming blocks */
 };
 
@@ -270 +280 @@
     tr_bool         haveFastSet;
 
+    uint8_t         state;
     uint8_t         ut_pex_id;
     uint16_t        pexCount;
@@ -1268 +1279 @@
 
     tr_peerIoReadUint32( msgs->peer->io, inbuf, &len );
-    msgs->incoming.length = len;
 
     if( len == 0 ) /* peer sent us a keepalive message */
         dbgmsg( msgs, "got KeepAlive" );
+    else
+    {
+        msgs->incoming.length = len;
+        msgs->state = AWAITING_BT_ID;
+    }
 
     return READ_NOW;
+}
+
+static int readBtMessage( tr_peermsgs *     msgs,
+                          struct evbuffer * inbuf,
+                          size_t            inlen );
+
+static int
+readBtId( tr_peermsgs *     msgs,
+          struct evbuffer * inbuf,
+          size_t            inlen )
+{
+    uint8_t id;
+
+    if( inlen < sizeof( uint8_t ) )
+        return READ_LATER;
+
+    tr_peerIoReadUint8( msgs->peer->io, inbuf, &id );
+    msgs->incoming.id = id;
+
+    if( id == BT_PIECE )
+    {
+        msgs->state = AWAITING_BT_PIECE;
+        return READ_NOW;
+    }
+    else if( msgs->incoming.length != 1 )
+    {
+        msgs->state = AWAITING_BT_MESSAGE;
+        return READ_NOW;
+    }
+    else return readBtMessage( msgs, inbuf, inlen - 1 );
 }
 
@@ -1362 +1407 @@
              size_t           * setme_piece_bytes_read )
 {
-    struct peer_request req;
+    struct peer_request * req = &msgs->incoming.blockReq;
 
     assert( EVBUFFER_LENGTH( inbuf ) >= inlen );
     dbgmsg( msgs, "In readBtPiece" );
 
-    tr_peerIoReadUint32( msgs->peer->io, inbuf, &req.index );
-    tr_peerIoReadUint32( msgs->peer->io, inbuf, &req.offset );
-    req.length = msgs->incoming.length - 9;
-    dbgmsg( msgs, "got incoming block header %u:%u->%u", req.index, req.offset, req.length );
-
-    { 
+    if( !req->length )
+    {
+        if( inlen < 8 )
+            return READ_LATER;
+
+        tr_peerIoReadUint32( msgs->peer->io, inbuf, &req->index );
+        tr_peerIoReadUint32( msgs->peer->io, inbuf, &req->offset );
+        req->length = msgs->incoming.length - 9;
+        dbgmsg( msgs, "got incoming block header %u:%u->%u", req->index, req->offset, req->length );
+        return READ_NOW;
+    }
+    else
+    {
         int err;
 
-        /* decrypt the whole block in one go */
-        evbuffer_expand( msgs->incoming.block, req.length );
-        tr_peerIoReadBytes( msgs->peer->io, inbuf, EVBUFFER_DATA( msgs->incoming.block ), req.length );
-        EVBUFFER_LENGTH( msgs->incoming.block ) += req.length;
-
-        fireClientGotData( msgs, req.length, TRUE );
-        *setme_piece_bytes_read += req.length;
-        dbgmsg( msgs, "got block %u:%u->%u", req.index, req.offset, req.length );
-        assert( EVBUFFER_LENGTH( msgs->incoming.block ) == req.length );
+        /* read in another chunk of data */
+        const size_t nLeft = req->length - EVBUFFER_LENGTH( msgs->incoming.block );
+        size_t n = MIN( nLeft, inlen );
+        size_t i = n;
+
+        while( i > 0 )
+        {
+            uint8_t buf[MAX_STACK_ARRAY_SIZE];
+            const size_t thisPass = MIN( i, sizeof( buf ) );
+            tr_peerIoReadBytes( msgs->peer->io, inbuf, buf, thisPass );
+            evbuffer_add( msgs->incoming.block, buf, thisPass );
+            i -= thisPass;
+        }
+
+        fireClientGotData( msgs, n, TRUE );
+        *setme_piece_bytes_read += n;
+        dbgmsg( msgs, "got %zu bytes for block %u:%u->%u ... %d remain",
+               n, req->index, req->offset, req->length,
+               (int)( req->length - EVBUFFER_LENGTH( msgs->incoming.block ) ) );
+        if( EVBUFFER_LENGTH( msgs->incoming.block ) < req->length )
+            return READ_LATER;
 
         /* we've got the whole block ... process it */
-        err = clientGotBlock( msgs, EVBUFFER_DATA( msgs->incoming.block ), &req );
+        err = clientGotBlock( msgs, EVBUFFER_DATA( msgs->incoming.block ), req );
 
         /* cleanup */
         evbuffer_drain( msgs->incoming.block, EVBUFFER_LENGTH( msgs->incoming.block ) );
+        req->length = 0;
+        msgs->state = AWAITING_BT_LENGTH;
         if( !err )
             return READ_NOW;
@@ -1400 +1466 @@
 
 static int
-readBtMessage( tr_peermsgs * msgs, struct evbuffer * inbuf, size_t inlen, size_t * piece )
-{
-    int           ret = READ_NOW;
-    uint8_t       id;
+readBtMessage( tr_peermsgs * msgs, struct evbuffer * inbuf, size_t inlen )
+{
     uint32_t      ui32;
     uint32_t      msglen = msgs->incoming.length;
+    const uint8_t id = msgs->incoming.id;
     const size_t  startBufLen = EVBUFFER_LENGTH( inbuf );
     const tr_bool fext = tr_peerIoSupportsFEXT( msgs->peer->io );
 
+    --msglen; /* id length */
+
     if( inlen < msglen )
         return READ_LATER;
 
-    tr_peerIoReadUint8( msgs->peer->io, inbuf, &id );
-
     dbgmsg( msgs, "got BT id %d, len %d, buffer size is %zu", (int)id, (int)msglen, inlen );
 
-    if( !messageLengthIsCorrect( msgs, id, msglen ) )
+    if( !messageLengthIsCorrect( msgs, id, msglen + 1 ) )
     {
         dbgmsg( msgs, "bad packet - BT message #%d with a length of %d", (int)id, (int)msglen );
@@ -1422 +1487 @@
         return READ_ERR;
     }
-
-    --msglen;
 
     switch( id )
@@ -1495 +1558 @@
 
         case BT_PIECE:
-            ret = readBtPiece( msgs, inbuf, msglen, piece );
+            assert( 0 ); /* handled elsewhere! */
             break;
 
@@ -1574 +1637 @@
     }
 
-    assert( EVBUFFER_LENGTH( inbuf ) == startBufLen - msglen - 1 );
-
-    msgs->incoming.length = 0;
-    return ret;
+    assert( msglen + 1 == msgs->incoming.length );
+    assert( EVBUFFER_LENGTH( inbuf ) == startBufLen - msglen );
+
+    msgs->state = AWAITING_BT_LENGTH;
+    return READ_NOW;
 }
 
@@ -1678 +1742 @@
 
     if( !inlen )
-        return READ_LATER;
-
-    /* Incoming data is processed in two stages. First the length is read
-     * and then readBtMessage() waits until all the data has arrived in
-     * the input buffer before starting to parse it */
-    if( msgs->incoming.length == 0 )
-        ret = readBtLength ( msgs, in, inlen );
-    else
-        ret = readBtMessage( msgs, in, inlen, piece );
+    {
+        ret = READ_LATER;
+    }
+    else if( msgs->state == AWAITING_BT_PIECE )
+    {
+        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 );
+    }
 
     /* log the raw data that was read */
@@ -2166 +2242 @@
     m->peer->peerIsInterested = 0;
     m->peer->have = tr_bitfieldNew( torrent->info.pieceCount );
+    m->state = AWAITING_BT_LENGTH;
     m->pexTimer = tr_timerNew( m->session, pexPulse, m, PEX_INTERVAL );
     m->outMessages = evbuffer_new( );