Changeset 7402

Timestamp:

Dec 15, 2008, 9:22:08 PM (12 years ago)

Author:

charles

Message:

(trunk libT) probable fix for the "greedy peer" bug

Location:

trunk/libtransmission

Files:

• bandwidth.c (modified) (5 diffs)
• iobuf.c (modified) (4 diffs)
• iobuf.h (modified) (2 diffs)
• peer-io.c (modified) (1 diff)
• peer-mgr.c (modified) (3 diffs)

trunk/libtransmission/bandwidth.c

@@ -250 +250 @@
                    tr_direction    dir,
                    int             period_msec,
-                   tr_ptrArray   * addme_buffers )
+                   tr_ptrArray   * iobuf_pool )
 {
     assert( isBandwidth( b ) );
@@ -269 +269 @@
     }
 
-    /* notify the io buffers that there's more bandwidth available */
     {
         int i;
         const int n = tr_ptrArraySize( b->iobufs );
         for( i=0; i<n; ++i )
-            tr_ptrArrayAppend( addme_buffers, tr_ptrArrayNth( b->iobufs, i ) );
+            tr_ptrArrayAppend( iobuf_pool, tr_ptrArrayNth( b->iobufs, i ) );
     }
 
@@ -287 +286 @@
         struct tr_bandwidth ** children = (struct tr_bandwidth**) tr_ptrArrayPeek( b->children, &n );
         for( i=0; i<n; ++i )
-            tr_bandwidthAllocate( children[i], dir, period_msec );
+            allocateBandwidth( children[i], dir, period_msec, iobuf_pool );
     }
 }
@@ -299 +298 @@
     tr_ptrArray * tmp;
     struct tr_iobuf ** buffers;
-    const short what = dir==TR_UP ? EV_WRITE : EV_READ;
+    const size_t chunkSize = 1024; /* arbitrary */
 
     tmp = tr_ptrArrayNew( );
@@ -305 +304 @@
     buffers = (struct tr_iobuf**) tr_ptrArrayPeek( tmp, &n );
 
-    /* notify the io buffers in a random order s.t. no
-       particular peer gets to hog all the bandwidth */
-    while( n > 0 ) {
-        const int i = tr_cryptoRandInt( n );
-        tr_iobuf_enable( buffers[i], what );
-        buffers[i] = buffers[n-1];
-        --n;
-    }
-
+    /* loop through all the peers, reading and writing in small chunks,
+     * until we run out of bandwidth or peers. we do it this way to 
+     * prevent one peer from using up all the bandwidth */
+    while( n > 0 )
+    {
+        int i;
+        for( i=0; i<n; )
+        {
+            int byteCount;
+            if( dir == TR_UP )
+                byteCount = tr_iobuf_flush_output_buffer( buffers[i], chunkSize );
+            else
+                byteCount = tr_iobuf_tryread( buffers[i], chunkSize );
+            if( byteCount == (int)chunkSize )
+                ++i;
+            else
+                buffers[i] = buffers[--n];
+        }
+    }
+
+    /* cleanup */
     tr_ptrArrayFree( tmp, NULL );
 }

trunk/libtransmission/iobuf.c

@@ -70 +70 @@
     int magicNumber;
 
+    int fd;
+
     int timeout_read;  /* in seconds */
     int timeout_write; /* in seconds */
@@ -89 +91 @@
 
 static int
-tr_evbuffer_write( struct evbuffer *buffer, int fd, size_t maxlen )
-{
-    int n = MIN( EVBUFFER_LENGTH( buffer ), maxlen );
+tr_evbuffer_write( struct evbuffer *buffer, int fd, size_t howmuch )
+{
+    int n = MIN( EVBUFFER_LENGTH( buffer ), howmuch );
 
 #ifdef WIN32
@@ -107 +109 @@
 }
 
+int
+tr_iobuf_flush_output_buffer( struct tr_iobuf * b, size_t howmuch )
+{
+    int res;
+
+    assert( isBuf( b ) );
+
+    howmuch = tr_bandwidthClamp( b->bandwidth, TR_UP, howmuch );
+    howmuch = MIN( howmuch, EVBUFFER_LENGTH( b->output ) );
+
+    res = howmuch ? tr_evbuffer_write( b->output, b->fd, howmuch ) : 0;
+
+    if( ( res > 0 ) && ( b->writecb != NULL ) )
+        (*b->writecb)( b, (size_t)res, b->cbarg );
+
+    if( ( res < 0 ) && ( b->errorcb != NULL ) && ( errno != EAGAIN && errno != EINTR && errno != EINPROGRESS ) )
+        (*b->errorcb)( b, (short)res, b->cbarg );
+
+    return res;
+}
+
+int
+tr_iobuf_tryread( struct tr_iobuf * b, size_t howmuch )
+{
+    int res;
+
+    assert( isBuf( b ) );
+
+    howmuch = tr_bandwidthClamp( b->bandwidth, TR_DOWN, howmuch );
+
+    res = howmuch ? evbuffer_read( b->input, b->fd, howmuch ) : 0;
+
+    if( ( res > 0 ) && ( b->readcb != NULL ) )
+        (*b->readcb)( b, (size_t)res, b->cbarg );
+
+    if( ( res < 0 ) && ( b->errorcb != NULL ) && ( errno != EAGAIN && errno != EINTR ) )
+        (*b->errorcb)( b, (short)res, b->cbarg );
+
+    return res;
+}
+
+
 static int
 tr_iobuf_add(struct event *ev, int timeout)
@@ -259 +303 @@
     b = tr_new0( struct tr_iobuf, 1 );
     b->magicNumber = MAGIC_NUMBER;
+    b->fd = fd;
     b->session = session;
     b->bandwidth = bandwidth;

trunk/libtransmission/iobuf.h

@@ -45 +45 @@
  * 2. the up/down speeds are directly constrained by our `bandwidth' object
  * 3. the implementation is hidden in the .c file
+ *
+ * 4. a late addition nasty hack to read/write on demand, called from
+ *    bandwidth. this actually seems to make a lot of this class redundant
+ *    and probably should be refactored.
  */
 struct tr_iobuf;
@@ -105 +109 @@
 int tr_iobuf_disable( struct tr_iobuf * iobuf, short event );
 
+int tr_iobuf_flush_output_buffer( struct tr_iobuf * iobuf, size_t max );
+
+int tr_iobuf_tryread( struct tr_iobuf * iobuf, size_t max );
+
 #endif

trunk/libtransmission/peer-io.c

@@ -544 +544 @@
 {
     /* this is all kind of arbitrary, but what seems to work well is
-     * being large enough to hold the next 15 seconds' worth of input,
-     * or two and a half blocks, whichever is bigger.
+     * being large enough to hold the next 20 seconds' worth of input,
+     * or a few blocks, whichever is bigger.
      * It's okay to tweak this as needed */
     const double maxBlockSize = 16 * 1024; /* 16 KiB is from BT spec */
     const double currentSpeed = tr_bandwidthGetPieceSpeed( io->bandwidth, TR_UP );
     const double period = 20; /* arbitrary */
-    return MAX( maxBlockSize*2.5, currentSpeed*1024*period );
+    return MAX( maxBlockSize*5.5, currentSpeed*1024*period );
 }
 

trunk/libtransmission/peer-mgr.c

@@ -1133 +1133 @@
                 addStrike( t, peer );
                 peer->doPurge = 1;
+                tordbg( t, "setting doPurge because we got an EINVAL error" );
             }
             else if( ( e->err == ERANGE )
@@ -1140 +1141 @@
                 /* some protocol error from the peer */
                 peer->doPurge = 1;
+                tordbg( t, "setting doPurge because we got an ERANGE, EMSGSIZE, or ENOTCONN error" );
             }
             else /* a local error, such as an IO error */
@@ -2290 +2292 @@
             else
                 ++atom->numFails;
+             
+            fprintf( stderr, "removing bad peer %s\n", tr_peerIoGetAddrStr( peer->io ) );
             tordbg( t, "removing bad peer %s", tr_peerIoGetAddrStr( peer->io ) );
             removePeer( t, peer );