Changeset 5980

Timestamp:

May 31, 2008, 12:16:26 AM (13 years ago)

Author:

charles

Message:

experimental message batcher as discussed in ​http://forum.transmissionbt.com/viewtopic.php?p=24251#24251

File:

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

trunk/libtransmission/peer-msgs.c

@@ -65 +65 @@
 
     /* idle seconds before we send a keepalive */
-    KEEPALIVE_INTERVAL_SECS = 90,
+    KEEPALIVE_INTERVAL_SECS = 100,
 
     PEX_INTERVAL            = (90 * 1000), /* msec between sendPex() calls */
@@ -85 +85 @@
     /* 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 = 120
+    SENT_REQUEST_TTL_SECS = 120,
+
+    /* used in lowering the outMessages queue period */
+    IMMEDIATE_PRIORITY_INTERVAL_SECS = 0,
+    HIGH_PRIORITY_INTERVAL_SECS = 5,
+    LOW_PRIORITY_INTERVAL_SECS = 30
 };
 
@@ -268 +273 @@
     time_t clientSentPexAt;
     time_t clientSentAnythingAt;
+
+    /* when we started batching the outMessages */
+    time_t outMessagesBatchedAt;
+
+    /* how long the outMessages batch should be allowed to grow before
+     * it's flushed -- some messages (like requests >:) should be sent
+     * very quickly; others aren't as urgent. */
+    int outMessagesBatchPeriod;
     
     tr_bitfield * peerAllowedPieces;
@@ -316 +329 @@
 
 static void
+pokeBatchPeriod( tr_peermsgs * msgs, int interval )
+{
+    if( msgs->outMessagesBatchPeriod > interval )
+    {
+        msgs->outMessagesBatchPeriod = interval;
+        dbgmsg( msgs, "lowering batch interval to %d seconds", interval );
+    }
+}
+
+static void
 protocolSendRequest( tr_peermsgs * msgs, const struct peer_request * req )
 {
@@ -327 +350 @@
     tr_peerIoWriteUint32( io, out, req->offset );
     tr_peerIoWriteUint32( io, out, req->length );
+    pokeBatchPeriod( msgs, HIGH_PRIORITY_INTERVAL_SECS );
 }
 
@@ -341 +365 @@
     tr_peerIoWriteUint32( io, out, req->offset );
     tr_peerIoWriteUint32( io, out, req->length );
+    pokeBatchPeriod( msgs, IMMEDIATE_PRIORITY_INTERVAL_SECS );
 }
 
@@ -353 +378 @@
     tr_peerIoWriteUint8 ( io, out, BT_HAVE );
     tr_peerIoWriteUint32( io, out, index );
+    pokeBatchPeriod( msgs, LOW_PRIORITY_INTERVAL_SECS );
 }
 
@@ -364 +390 @@
     tr_peerIoWriteUint32( io, out, sizeof(uint8_t) );
     tr_peerIoWriteUint8 ( io, out, choke ? BT_CHOKE : BT_UNCHOKE );
+    pokeBatchPeriod( msgs, IMMEDIATE_PRIORITY_INTERVAL_SECS );
 }
 
@@ -497 +524 @@
     tr_peerIoWriteUint8 ( msgs->io, msgs->outMessages,
                    weAreInterested ? BT_INTERESTED : BT_NOT_INTERESTED );
+    pokeBatchPeriod( msgs, HIGH_PRIORITY_INTERVAL_SECS );
 }
 
@@ -596 +624 @@
         tr_peerIoWriteUint32( msgs->io, msgs->outMessages, offset );
         tr_peerIoWriteUint32( msgs->io, msgs->outMessages, length );
+        pokeBatchPeriod( msgs, LOW_PRIORITY_INTERVAL_SECS );
     }
 }
@@ -625 +654 @@
         tr_peerIoWriteUint8( msgs->io, msgs->outMessages, BT_ALLOWED_FAST );
         tr_peerIoWriteUint32( msgs->io, msgs->outMessages, pieceIndex );
+        pokeBatchPeriod( msgs, LOW_PRIORITY_INTERVAL_SECS );
     }
 }
@@ -1571 +1601 @@
     dbgmsg( msgs, "sending a keepalive message" );
     tr_peerIoWriteUint32( msgs->io, msgs->outMessages, 0 );
+    pokeBatchPeriod( msgs, IMMEDIATE_PRIORITY_INTERVAL_SECS );
 }
 
@@ -1657 +1688 @@
     {
         struct peer_request req;
-
-        if(( EVBUFFER_LENGTH( msgs->outMessages ) ))
+        int haveMessages = EVBUFFER_LENGTH( msgs->outMessages ) != 0;
+
+        if( haveMessages && !msgs->outMessagesBatchedAt ) /* fresh batch */
         {
-            dbgmsg( msgs, "flushing outMessages..." );
+            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 )
@@ -1687 +1727 @@
             tr_free( buf );
         }
-        else if( ( now - msgs->clientSentAnythingAt ) > KEEPALIVE_INTERVAL_SECS )
+        else if(    ( !haveMessages )
+                 && ( now - msgs->clientSentAnythingAt ) > KEEPALIVE_INTERVAL_SECS )
         {
             sendKeepalive( msgs );
@@ -1717 +1758 @@
     tr_peerIoWriteUint8 ( msgs->io, out, BT_BITFIELD );
     tr_peerIoWriteBytes ( msgs->io, out, bitfield->bits, bitfield->len );
+    pokeBatchPeriod( msgs, IMMEDIATE_PRIORITY_INTERVAL_SECS );
 }
 
@@ -1839 +1881 @@
         tr_peerIoWriteUint8 ( msgs->io, msgs->outMessages, msgs->ut_pex_id );
         tr_peerIoWriteBytes ( msgs->io, msgs->outMessages, benc, bencLen );
+        pokeBatchPeriod( msgs, IMMEDIATE_PRIORITY_INTERVAL_SECS );
 
         /* cleanup */
@@ -1890 +1933 @@
     m->pexTimer = tr_timerNew( m->handle, pexPulse, m, PEX_INTERVAL );
     m->outMessages = evbuffer_new( );
+    m->outMessagesBatchedAt = 0;
+    m->outMessagesBatchPeriod = LOW_PRIORITY_INTERVAL_SECS;
     m->incoming.block = evbuffer_new( );
     m->outBlock = evbuffer_new( );