Changeset 7154

Timestamp:

Nov 25, 2008, 9:35:17 PM (12 years ago)

Author:

charles

Message:

(libT) yet another stab at getting bandwidth management under control. this version may suck less than previous attempts. It also breaks the mac build until someone adds iobuf.[ch] to xcode...

Location:

trunk/libtransmission

Files:

• Makefile.am (modified) (2 diffs)
• bandwidth.c (modified) (6 diffs)
• bandwidth.h (modified) (2 diffs)
• handshake.c (modified) (4 diffs)
• iobuf.c (added)
• iobuf.h (added)
• net.c (modified) (2 diffs)
• peer-io.c (modified) (29 diffs)
• peer-io.h (modified) (3 diffs)
• peer-mgr-private.h (modified) (2 diffs)
• peer-mgr.c (modified) (14 diffs)
• peer-msgs.c (modified) (6 diffs)
• session.c (modified) (5 diffs)
• session.h (modified) (3 diffs)
• torrent.c (modified) (5 diffs)
• torrent.h (modified) (2 diffs)

trunk/libtransmission/Makefile.am

@@ -25 +25 @@
     handshake.c \
     inout.c \
+    iobuf.c \
     json.c \
     JSON_parser.c \
@@ -70 +71 @@
     handshake.h \
     inout.h \
+    iobuf.h \
     json.h \
     JSON_parser.h \

trunk/libtransmission/bandwidth.c

@@ -11 +11 @@
  */
 
+#include <assert.h>
 #include <limits.h>
+
+#include "event.h"
+
 #include "transmission.h"
 #include "bandwidth.h"
+#include "iobuf.h"
+#include "ptrarray.h"
 #include "utils.h"
 
@@ -25 +31 @@
     INTERVAL_MSEC = HISTORY_MSEC,
     GRANULARITY_MSEC = 250,
-    HISTORY_SIZE = ( INTERVAL_MSEC / GRANULARITY_MSEC )
+    HISTORY_SIZE = ( INTERVAL_MSEC / GRANULARITY_MSEC ),
+    MAGIC_NUMBER = 43143
 };
 
@@ -76 +83 @@
 ******/
 
-struct tr_bandwidth
+struct tr_band
 {
     unsigned int isLimited : 1;
-
+    unsigned int honorParentLimits : 1;
     size_t bytesLeft;
-
+    double desiredSpeed;
     struct bratecontrol raw;
     struct bratecontrol piece;
-
+};
+
+struct tr_bandwidth
+{
+    struct tr_band band[2];
+    struct tr_bandwidth * parent;
+    int magicNumber;
     tr_session * session;
+    tr_ptrArray * children; /* struct tr_bandwidth */
+    tr_ptrArray * iobufs; /* struct tr_iobuf */
 };
 
@@ -92 +107 @@
 ***/
 
+static int
+comparePointers( const void * a, const void * b )
+{
+    return a - b;
+}
+
+static int
+isBandwidth( const tr_bandwidth * b )
+{
+    return ( b != NULL ) && ( b->magicNumber == MAGIC_NUMBER );
+}
+
+static int
+isDirection( const tr_direction dir )
+{
+    return ( dir == TR_UP ) || ( dir == TR_DOWN );
+}
+
+/***
+****
+***/
+
 tr_bandwidth*
-tr_bandwidthNew( tr_session * session )
+tr_bandwidthNew( tr_session * session, tr_bandwidth * parent )
 {
     tr_bandwidth * b = tr_new0( tr_bandwidth, 1 );
     b->session = session;
+    b->children = tr_ptrArrayNew( );
+    b->iobufs = tr_ptrArrayNew( );
+    b->magicNumber = MAGIC_NUMBER;
+    b->band[TR_UP].honorParentLimits = 1;
+    b->band[TR_DOWN].honorParentLimits = 1;
+    tr_bandwidthSetParent( b, parent );
     return b;
 }
@@ -103 +146 @@
 tr_bandwidthFree( tr_bandwidth * b )
 {
+    assert( isBandwidth( b ) );
+
+    tr_bandwidthSetParent( b, NULL );
+    tr_ptrArrayFree( b->iobufs, NULL );
+    tr_ptrArrayFree( b->children, NULL );
+    b->magicNumber = 0xDEAD;
     tr_free( b );
 }
@@ -111 +160 @@
 
 void
+tr_bandwidthSetParent( tr_bandwidth  * b,
+                       tr_bandwidth  * parent )
+{
+    assert( isBandwidth( b ) );
+    assert( b != parent );
+
+    if( b->parent )
+    {
+        assert( isBandwidth( b->parent ) );
+
+        tr_ptrArrayRemoveSorted( b->parent->children, b, comparePointers );
+        b->parent= NULL;
+    }
+
+    if( parent )
+    {
+        assert( isBandwidth( parent ) );
+        assert( parent->parent != b );
+
+        tr_ptrArrayInsertSorted( parent->children, b, comparePointers );
+        b->parent = parent;
+    }
+}
+
+void
+tr_bandwidthHonorParentLimits( tr_bandwidth  * b,
+                               tr_direction    dir,
+                               int             honorParentLimits )
+{
+    assert( isBandwidth( b ) );
+    assert( isDirection( dir ) );
+
+    b->band[dir].honorParentLimits = honorParentLimits != 0;
+}
+
+/***
+****
+***/
+
+void
+tr_bandwidthSetDesiredSpeed( tr_bandwidth  * b,
+                             tr_direction    dir,
+                             double          desiredSpeed )
+{
+    assert( isBandwidth( b ) );
+    assert( isDirection( dir ) );
+
+    b->band[dir].desiredSpeed = desiredSpeed; 
+}
+
+double
+tr_bandwidthGetDesiredSpeed( const tr_bandwidth  * b,
+                             tr_direction          dir )
+{
+    assert( isBandwidth( b ) );
+    assert( isDirection( dir ) );
+
+    return b->band[dir].desiredSpeed;
+}
+
+void
 tr_bandwidthSetLimited( tr_bandwidth  * b,
-                        size_t          bytesLeft )
-{
-    b->isLimited = 1;
-    b->bytesLeft = bytesLeft;
-}
-
-void
-tr_bandwidthSetUnlimited( tr_bandwidth * b )
-{
-    b->isLimited = 0;
-}
+                        tr_direction    dir,
+                        int             isLimited )
+{
+    assert( isBandwidth( b ) );
+    assert( isDirection( dir ) );
+
+    b->band[dir].isLimited = isLimited != 0;
+}
+
+int
+tr_bandwidthIsLimited( const tr_bandwidth  * b,
+                       tr_direction          dir )
+{
+    assert( isBandwidth( b ) );
+    assert( isDirection( dir ) );
+
+    return b->band[dir].isLimited != 0;
+}
+
+#if 0
+#define DEBUG_DIRECTION TR_DOWN
+#endif
+
+void
+tr_bandwidthAllocate( tr_bandwidth  * b,
+                      tr_direction    dir,
+                      int             period_msec )
+{
+    const double currentSpeed = tr_bandwidthGetPieceSpeed( b, dir ); /* KiB/s */
+    const double desiredSpeed = b->band[dir].desiredSpeed;           /* KiB/s */
+    const double seconds_per_pulse = period_msec / 1000.0;
+    const double current_bytes_per_pulse = currentSpeed * 1024.0 * seconds_per_pulse;
+    const double desired_bytes_per_pulse = desiredSpeed * 1024.0 * seconds_per_pulse;
+    const double pulses_per_history = (double)HISTORY_MSEC / period_msec;
+    const double min = desired_bytes_per_pulse * 0.90;
+    const double max = desired_bytes_per_pulse * 1.50;
+    const double next_pulse_bytes = desired_bytes_per_pulse * ( pulses_per_history + 1 )
+                                  - ( current_bytes_per_pulse * pulses_per_history );
+    double clamped;
+
+    /* clamp the return value to lessen oscillation */
+    clamped = next_pulse_bytes;
+    clamped = MAX( clamped, min );
+    clamped = MIN( clamped, max );
+    b->band[dir].bytesLeft = clamped;
+
+#ifdef DEBUG_DIRECTION
+if( dir == DEBUG_DIRECTION )
+fprintf( stderr, "bandwidth %p currentSpeed(%5.2f) desiredSpeed(%5.2f), allocating %5.2f (unclamped: %5.2f)\n",
+         b, currentSpeed, desiredSpeed,
+         clamped/1024.0, next_pulse_bytes/1024.0 );
+#endif
+
+    /* notify the io buffers that there's more bandwidth available */
+    if( !b->band[dir].isLimited || ( clamped > 0 ) ) {
+        int i, n=0;
+        short what = dir==TR_UP ? EV_WRITE : EV_READ;
+        struct tr_iobuf ** iobufs = (struct tr_iobuf**) tr_ptrArrayPeek( b->iobufs, &n );
+#ifdef DEBUG_DIRECTION
+if( dir == DEBUG_DIRECTION )
+fprintf( stderr, "bandwidth %p has %d iobufs\n", b, n );
+#endif
+        for( i=0; i<n; ++i )
+            tr_iobuf_enable( iobufs[i], what );
+    }
+
+    /* all children should reallocate too */
+    if( 1 ) {
+        int i, n=0;
+        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 );
+    }
+}
+
+/***
+****
+***/
+
+void
+tr_bandwidthAddBuffer( tr_bandwidth        * b,
+                       struct tr_iobuf     * iobuf )
+{
+    assert( isBandwidth( b ) );
+    assert( iobuf );
+
+    tr_ptrArrayInsertSorted( b->iobufs, iobuf, comparePointers );
+}
+
+void
+tr_bandwidthRemoveBuffer( tr_bandwidth        * b,
+                          struct tr_iobuf     * iobuf )
+{
+    assert( isBandwidth( b ) );
+    assert( iobuf );
+
+    tr_ptrArrayRemoveSorted( b->iobufs, iobuf, comparePointers );
+}
+
+/***
+****
+***/
 
 size_t
 tr_bandwidthClamp( const tr_bandwidth  * b,
+                   tr_direction          dir,
                    size_t                byteCount )
 {
-/* const size_t n = byteCount; */
-
-    if( b && b->isLimited )
-        byteCount = MIN( byteCount, b->bytesLeft );
-
-/* if( n != byteCount ) fprintf( stderr, "%p: %zu clamped to %zu\n", b, n, byteCount ); */
+    assert( isBandwidth( b ) );
+    assert( isDirection( dir ) );
+
+    if( b )
+    {
+        if( b->band[dir].isLimited )
+            byteCount = MIN( byteCount, b->band[dir].bytesLeft );
+
+        if( b->parent && b->band[dir].honorParentLimits )
+            byteCount = tr_bandwidthClamp( b->parent, dir, byteCount );
+    }
+
     return byteCount;
 }
 
-/***
-****
-***/
-
 double
-tr_bandwidthGetRawSpeed( const tr_bandwidth * b )
-{
-    return getSpeed( &b->raw );
+tr_bandwidthGetRawSpeed( const tr_bandwidth * b, tr_direction dir )
+{
+    assert( isBandwidth( b ) );
+    assert( isDirection( dir ) );
+
+    return getSpeed( &b->band[dir].raw );
 }
 
 double
-tr_bandwidthGetPieceSpeed( const tr_bandwidth * b UNUSED )
-{
-    return getSpeed( &b->piece );
+tr_bandwidthGetPieceSpeed( const tr_bandwidth * b, tr_direction dir )
+{
+    assert( isBandwidth( b ) );
+    assert( isDirection( dir ) );
+
+    return getSpeed( &b->band[dir].piece );
 }
 
 void
 tr_bandwidthUsed( tr_bandwidth  * b,
+                  tr_direction    dir,
                   size_t          byteCount,
                   int             isPieceData )
 {
-    if( b->isLimited && isPieceData )
-    {
-        b->bytesLeft -= MIN( b->bytesLeft, byteCount );
-        /* fprintf( stderr, "%p used %zu bytes ... %zu left\n", b, byteCount, b->bytesLeft ); */
-    }
-
-    bytesUsed( &b->raw, byteCount );
+    struct tr_band * band;
+
+    assert( isBandwidth( b ) );
+    assert( isDirection( dir ) );
+
+    band = &b->band[dir];
+
+    if( band->isLimited && isPieceData )
+        band->bytesLeft -= MIN( band->bytesLeft, byteCount );
+
+#ifdef DEBUG_DIRECTION
+if( ( dir == DEBUG_DIRECTION ) && band->isLimited && isPieceData )
+fprintf( stderr, "%p consumed %zu bytes of piece data... %zu left\n", b, byteCount, band->bytesLeft );
+#endif
+
+    bytesUsed( &band->raw, byteCount );
 
     if( isPieceData )
-        bytesUsed( &b->piece, byteCount );
-}
+        bytesUsed( &band->piece, byteCount );
+
+    if( b->parent != NULL )
+        tr_bandwidthUsed( b->parent, dir, byteCount, isPieceData );
+}

trunk/libtransmission/bandwidth.h

@@ -18 +18 @@
 #define TR_BANDWIDTH_H
 
+struct tr_iobuf;
+
+/**
+ * Bandwidth is an object for measuring and constraining bandwidth speeds.
+ *
+ * Bandwidth objects can be "stacked" so that a peer can be made to obey
+ * multiple constraints (for example, obeying the global speed limit and a
+ * per-torrent speed limit).
+ *
+ * HIERARCHY
+ *
+ *   Transmission's bandwidth hierarchy is a tree.
+ *   At the top is the global bandwidth object owned by tr_session.
+ *   Its children are per-torrent bandwidth objects owned by tr_torrent.
+ *   Underneath those are per-peer bandwidth objects owned by tr_peer.
+ *
+ *   tr_session also owns a tr_handshake's bandwidths, so that the handshake
+ *   I/O can be counted in the global raw totals.  When the handshake is done,
+ *   the bandwidth's ownership passes to a tr_peer.
+ * 
+ * MEASURING
+ *
+ *   When you ask a bandwidth object for its speed, it gives the speed of the
+ *   subtree underneath it as well.  So you can get Transmission's overall
+ *   speed by quering tr_session's bandwidth, per-torrent speeds by asking
+ *   tr_torrent's bandwidth, and per-peer speeds by asking tr_peer's bandwidth.
+ *
+ * CONSTRAINING
+ * 
+ *   Call tr_bandwidthAllocate() periodically.  tr_bandwidth knows its current
+ *   speed and will decide how many bytes to make available over the
+ *   user-specified period to reach the user-specified desired speed.
+ *   If appropriate, it notifies its iobufs that new bandwidth is available.
+ * 
+ *   tr_bandwidthAllocate() operates on the tr_bandwidth subtree, so usually 
+ *   you'll only need to invoke it for the top-level tr_session bandwidth.
+ *
+ *   The iobufs all have a pointer to their associated tr_bandwidth object,
+ *   and call tr_bandwidthClamp() before performing I/O to see how much 
+ *   bandwidth they can safely use.
+ */
 typedef struct tr_bandwidth tr_bandwidth;
 
@@ -24 +65 @@
 **/
 
-tr_bandwidth* tr_bandwidthNew           ( tr_session          * session );
+/** @brief create a new tr_bandwidth object */
+tr_bandwidth*
+         tr_bandwidthNew              ( tr_session          * session,
+                                        tr_bandwidth        * parent );
 
-void          tr_bandwidthFree          ( tr_bandwidth        * bandwidth );
+/** @brief destroy a tr_bandwidth object */
+void     tr_bandwidthFree             ( tr_bandwidth        * bandwidth );
+
+/******
+*******
+******/
 
 /**
-***
-**/
-
-void          tr_bandwidthSetLimited    ( tr_bandwidth        * bandwidth,
-                                          size_t                byteCount );
-
-void          tr_bandwidthSetUnlimited  ( tr_bandwidth        * bandwidth );
-
-size_t        tr_bandwidthClamp         ( const tr_bandwidth  * bandwidth,
-                                          size_t                byteCount );
+ * @brief Set the desired speed (in KiB/s) for this bandwidth subtree.
+ * @see tr_bandwidthAllocate
+ * @see tr_bandwidthGetDesiredSpeed
+ */
+void    tr_bandwidthSetDesiredSpeed   ( tr_bandwidth        * bandwidth,
+                                        tr_direction          direction,
+                                        double                desiredSpeed );
 
 /**
-***
-**/
+ * @brief Get the desired speed (in KiB/s) for ths bandwidth subtree.
+ * @see tr_bandwidthSetDesiredSpeed
+ */
+double  tr_bandwidthGetDesiredSpeed   ( const tr_bandwidth  * bandwidth,
+                                        tr_direction          direction );
 
-double        tr_bandwidthGetRawSpeed   ( const tr_bandwidth  * bandwidth );
+/**
+ * @brief Set whether or not this bandwidth should throttle its iobufs' speeds
+ */
+void    tr_bandwidthSetLimited        ( tr_bandwidth        * bandwidth,
+                                        tr_direction          direction,
+                                        int                   isLimited );
 
-double        tr_bandwidthGetPieceSpeed ( const tr_bandwidth  * bandwidth );
+/**
+ * @return nonzero if this bandwidth throttles its iobufs' speeds
+ */
+int     tr_bandwidthIsLimited         ( const tr_bandwidth  * bandwidth,
+                                        tr_direction          direction );
 
-void          tr_bandwidthUsed          ( tr_bandwidth        * bandwidth,
-                                          size_t                byteCount,
-                                          int                   isPieceData );
+/**
+ * @brief allocate the next period_msec's worth of bandwidth for the iobufs to consume
+ */
+void    tr_bandwidthAllocate          ( tr_bandwidth        * bandwidth,
+                                        tr_direction          direction,
+                                        int                   period_msec );
+
+/**
+ * @brief clamps byteCount down to a number that this bandwidth will allow to be consumed
+ */
+size_t  tr_bandwidthClamp             ( const tr_bandwidth  * bandwidth,
+                                        tr_direction          direction,
+                                        size_t                byteCount );
+
+/******
+*******
+******/
+
+/**
+ * @brief Get the raw total of bytes read or sent by this bandwidth subtree.
+ */
+double  tr_bandwidthGetRawSpeed       ( const tr_bandwidth  * bandwidth,
+                                        tr_direction          direction );
+
+/**
+ * @brief Get the number of piece data bytes read or sent by this bandwidth subtree.
+ */
+double  tr_bandwidthGetPieceSpeed     ( const tr_bandwidth  * bandwidth,
+                                        tr_direction          direction );
+
+/**
+ * @brief Notify the bandwidth object that some of its allocated bandwidth has been consumed.
+ * This is is usually invoked by the iobuf after a read or write.
+ */
+void    tr_bandwidthUsed              ( tr_bandwidth        * bandwidth,
+                                        tr_direction          direction,
+                                        size_t                byteCount,
+                                        int                   isPieceData );
+
+/******
+*******
+******/
+
+void    tr_bandwidthSetParent         ( tr_bandwidth        * bandwidth,
+                                        tr_bandwidth        * parent );
+
+/**
+ * Almost all the time we do want to honor a parents' bandwidth cap, so that
+ * (for example) a peer is constrained by a per-torrent cap and the global cap.
+ * But when we set a torrent's speed mode to TR_SPEEDLIMIT_UNLIMITED, then
+ * in that particular case we want to ignore the global speed limit...
+ */
+void    tr_bandwidthHonorParentLimits ( tr_bandwidth        * bandwidth,
+                                        tr_direction          direction,
+                                        int                   isEnabled );
+
+/******
+*******
+******/
+
+/**
+ * @brief add an iobuf to this bandwidth's list of iobufs.
+ * They will be notified when more bandwidth is made available for them to consume.
+ */
+void    tr_bandwidthAddBuffer         ( tr_bandwidth        * bandwidth,
+                                        struct tr_iobuf     * iobuf );
+
+/**
+ * @brief remove an iobuf from this bandwidth's list of iobufs.
+ */
+void    tr_bandwidthRemoveBuffer      ( tr_bandwidth        * bandwidth,
+                                        struct tr_iobuf     * iobuf );
 
 #endif

trunk/libtransmission/handshake.c

@@ -25 +25 @@
 #include "crypto.h"
 #include "handshake.h"
+#include "iobuf.h"
 #include "peer-io.h"
 #include "peer-mgr.h"
@@ -1029 +1030 @@
 
 static ReadState
-canRead( struct bufferevent * evin,
-         void *               arg )
-{
-    tr_handshake *    handshake = (tr_handshake *) arg;
-    struct evbuffer * inbuf = EVBUFFER_INPUT ( evin );
+canRead( struct tr_iobuf * iobuf, void * arg, size_t * piece )
+{
+    tr_handshake *    handshake = arg;
+    struct evbuffer * inbuf = tr_iobuf_input( iobuf );
     ReadState         ret;
     int               readyForMore = TRUE;
+
+    /* no piece data in handshake */
+    *piece = 0;
 
     dbgmsg( handshake, "handling canRead; state is [%s]",
@@ -1137 +1140 @@
 
 static void
-gotError( struct bufferevent * evbuf UNUSED,
-          short                      what,
-          void *                     arg )
+gotError( struct tr_iobuf  * iobuf UNUSED,
+          short              what,
+          void             * arg )
 {
     tr_handshake * handshake = (tr_handshake *) arg;
@@ -1180 +1183 @@
     tr_handshake * handshake;
 
-    tr_peerIoSetBandwidth( io, TR_UP, NULL );
-    tr_peerIoSetBandwidth( io, TR_DOWN, NULL );
-
     handshake = tr_new0( tr_handshake, 1 );
     handshake->io = io;

trunk/libtransmission/net.c

@@ -126 +126 @@
 setSndBuf( tr_session * session, int fd )
 {
-#if 0
     if( fd >= 0 )
     {
@@ -134 +133 @@
         setsockopt( fd, SOL_SOCKET, SO_RCVBUF, &rcvbuf, sizeof( rcvbuf ) );
     }
-#endif
 }
 

trunk/libtransmission/peer-io.c

@@ -29 +29 @@
 #include "bandwidth.h"
 #include "crypto.h"
+#include "iobuf.h"
 #include "list.h"
 #include "net.h"
@@ -35 +36 @@
 #include "utils.h"
 
+#define MAGIC_NUMBER 206745
 #define IO_TIMEOUT_SECS 8
 
@@ -79 +81 @@
 struct tr_peerIo
 {
-    unsigned int           isEncrypted               : 1;
-    unsigned int           isIncoming                : 1;
-    unsigned int           peerIdIsSet               : 1;
-    unsigned int           extendedProtocolSupported : 1;
-    unsigned int           fastPeersSupported        : 1;
-
-    uint8_t                encryptionMode;
-    uint8_t                timeout;
-    uint16_t               port;
-    int                    socket;
-
-    uint8_t                peerId[20];
-    time_t                 timeCreated;
-
-    tr_session           * session;
-
-    struct in_addr         in_addr;
-    struct bufferevent   * bufev;
-    struct evbuffer      * output;
-    tr_list              * output_datatypes; /* struct tr_datatype */
-
-    tr_can_read_cb         canRead;
-    tr_did_write_cb        didWrite;
-    tr_net_error_cb        gotError;
-    void *                 userData;
-
-    size_t                 bufferSize[2];
-
-    tr_bandwidth         * bandwidth[2];
-    tr_crypto            * crypto;
+    unsigned int             isEncrypted               : 1;
+    unsigned int             isIncoming                : 1;
+    unsigned int             peerIdIsSet               : 1;
+    unsigned int             extendedProtocolSupported : 1;
+    unsigned int             fastPeersSupported        : 1;
+
+    int                      magicNumber;
+
+    uint8_t                  encryptionMode;
+    uint8_t                  timeout;
+    uint16_t                 port;
+    int                      socket;
+
+    uint8_t                  peerId[20];
+    time_t                   timeCreated;
+
+    tr_session             * session;
+
+    struct in_addr           in_addr;
+    struct tr_iobuf        * iobuf;
+    tr_list                * output_datatypes; /* struct tr_datatype */
+
+    tr_can_read_cb           canRead;
+    tr_did_write_cb          didWrite;
+    tr_net_error_cb          gotError;
+    void *                   userData;
+
+    size_t                   bufferSize[2];
+
+    tr_bandwidth           * bandwidth;
+    tr_crypto              * crypto;
 };
-
-/**
-***
-**/
-
-static void
-adjustOutputBuffer( tr_peerIo * io )
-{
-    struct evbuffer * live = EVBUFFER_OUTPUT( io->bufev );
-    size_t curLive = EVBUFFER_LENGTH( live );
-    size_t maxLive = tr_bandwidthClamp( io->bandwidth[TR_UP], io->session->so_sndbuf );
-
-    if( ( curLive < maxLive ) && EVBUFFER_LENGTH( io->output ) )
-    {
-        size_t freeSpace = maxLive - curLive;
-        size_t n = MIN( freeSpace, EVBUFFER_LENGTH( io->output ) );
-        bufferevent_write( io->bufev, EVBUFFER_DATA( io->output ), n );
-        evbuffer_drain( io->output, n );
-        curLive += n;
-    }
-
-    io->bufferSize[TR_UP] = curLive;
-
-    if( curLive )
-        bufferevent_enable( io->bufev, EV_WRITE );
-
-    dbgmsg( io, "after adjusting the output buffer, its size is now %zu", curLive );
-}
-
-static void
-adjustInputBuffer( tr_peerIo * io )
-{
-    /* FIXME: the max read size probably needs to vary depending on the
-     * number of peers that we have connected...  1024 is going to force
-     * us way over the limit when there are lots of peers */
-    static const int maxBufSize = 1024;
-    const size_t n = tr_bandwidthClamp( io->bandwidth[TR_DOWN], maxBufSize );
-
-    if( !n )
-    {
-        dbgmsg( io, "disabling reads because we've hit our limit" );
-        bufferevent_disable( io->bufev, EV_READ );
-    }
-    else
-    {
-        dbgmsg( io, "enabling reading of %zu more bytes", n );
-        bufferevent_setwatermark( io->bufev, EV_READ, 0, n );
-        bufferevent_enable( io->bufev, EV_READ );
-    }
-
-    io->bufferSize[TR_DOWN] = EVBUFFER_LENGTH( EVBUFFER_INPUT( io->bufev ) );
-}
 
 /***
@@ -168 +119 @@
 
 static void
-didWriteWrapper( struct bufferevent * e,
-                 void *               vio )
+didWriteWrapper( struct tr_iobuf  * iobuf,
+                 size_t             bytes_transferred,
+                 void             * vio )
 {
     tr_peerIo *  io = vio;
-    const size_t len = EVBUFFER_LENGTH( EVBUFFER_OUTPUT( e ) );
-
-    dbgmsg( io, "didWrite... io->outputBufferSize was %zu, is now %zu",
-            io->bufferSize[TR_UP], len );
-
-    if( len < io->bufferSize[TR_UP] )
+
+    while( bytes_transferred )
     {
-        size_t payload = io->bufferSize[TR_UP] - len;
-
-        while( payload )
-        {
-            struct tr_datatype * next = io->output_datatypes->data;
-            const size_t chunk_length = MIN( next->length, payload );
-            const size_t n = addPacketOverhead( chunk_length );
-
-            if( io->didWrite )
-                io->didWrite( io, n, next->isPieceData, io->userData );
-
-            payload -= chunk_length;
-            next->length -= chunk_length;
-            if( !next->length )
-                tr_free( tr_list_pop_front( &io->output_datatypes ) );
-        }
+        struct tr_datatype * next = io->output_datatypes->data;
+        const size_t chunk_length = MIN( next->length, bytes_transferred );
+        const size_t n = addPacketOverhead( chunk_length );
+
+        tr_bandwidthUsed( io->bandwidth, TR_UP, n, next->isPieceData );
+
+        if( io->didWrite )
+            io->didWrite( io, n, next->isPieceData, io->userData );
+
+        bytes_transferred -= chunk_length;
+        next->length -= chunk_length;
+        if( !next->length )
+            tr_free( tr_list_pop_front( &io->output_datatypes ) );
     }
 
-    adjustOutputBuffer( io );
-
+    if( EVBUFFER_LENGTH( tr_iobuf_output( iobuf ) ) )
+        tr_iobuf_enable( io->iobuf, EV_WRITE );
 }
 
 static void
-canReadWrapper( struct bufferevent * e,
-                void *               vio )
+canReadWrapper( struct tr_iobuf  * iobuf,
+                size_t             bytes_transferred UNUSED,
+                void              * vio )
 {
     int          done = 0;
@@ -219 +165 @@
         while( !done && !err )
         {
-            const int ret = io->canRead( e, io->userData );
+            size_t piece = 0;
+            const size_t oldLen = EVBUFFER_LENGTH( tr_iobuf_input( iobuf ) );
+            const int ret = io->canRead( iobuf, io->userData, &piece );
+
+            if( ret != err )
+            {
+                const size_t used = oldLen - EVBUFFER_LENGTH( tr_iobuf_input( iobuf ) );
+                if( piece )
+                    tr_bandwidthUsed( io->bandwidth, TR_DOWN, piece, TRUE );
+                if( used != piece )
+                    tr_bandwidthUsed( io->bandwidth, TR_DOWN, used - piece, FALSE );
+            }
 
             switch( ret )
             {
                 case READ_NOW:
-                    if( EVBUFFER_LENGTH( e->input ) )
+                    if( EVBUFFER_LENGTH( tr_iobuf_input( iobuf )))
                         continue;
                     done = 1;
@@ -241 +198 @@
         tr_globalUnlock( session );
     }
-
-    if( !err )
-        adjustInputBuffer( io );
 }
 
 static void
-gotErrorWrapper( struct bufferevent * e,
-                 short                what,
-                 void *               userData )
+gotErrorWrapper( struct tr_iobuf  * iobuf,
+                 short              what,
+                 void             * userData )
 {
     tr_peerIo * c = userData;
 
     if( c->gotError )
-        c->gotError( e, what, c->userData );
+        c->gotError( iobuf, what, c->userData );
 }
 
@@ -264 +218 @@
 bufevNew( tr_peerIo * io )
 {
-    io->bufev = bufferevent_new( io->socket,
-                                 canReadWrapper,
-                                 didWriteWrapper,
-                                 gotErrorWrapper,
-                                 io );
-
-    /* tell libevent to call didWriteWrapper after every write,
-     * not just when the write buffer is empty */
-    bufferevent_setwatermark( io->bufev, EV_WRITE, INT_MAX, 0 );
-
-    bufferevent_settimeout( io->bufev, io->timeout, io->timeout );
-
-    bufferevent_enable( io->bufev, EV_READ | EV_WRITE );
+    io->iobuf = tr_iobuf_new( io->session,
+                              io->bandwidth,
+                              io->socket,
+                              EV_READ | EV_WRITE,
+                              canReadWrapper,
+                              didWriteWrapper,
+                              gotErrorWrapper,
+                              io );
+
+    tr_iobuf_settimeout( io->iobuf, io->timeout, io->timeout );
+}
+
+static int
+isPeerIo( const tr_peerIo * io )
+{
+    return ( io != NULL ) && ( io->magicNumber == MAGIC_NUMBER );
 }
 
@@ -293 +250 @@
 
     io = tr_new0( tr_peerIo, 1 );
+    io->magicNumber = MAGIC_NUMBER;
     io->crypto = tr_cryptoNew( torrentHash, isIncoming );
     io->session = session;
@@ -301 +259 @@
     io->timeout = IO_TIMEOUT_SECS;
     io->timeCreated = time( NULL );
-    io->output = evbuffer_new( );
     bufevNew( io );
+    tr_peerIoSetBandwidth( io, session->bandwidth );
     return io;
 }
@@ -346 +304 @@
     tr_peerIo * io = vio;
 
-    evbuffer_free( io->output );
-    bufferevent_free( io->bufev );
+    tr_peerIoSetBandwidth( io, NULL );
+    tr_iobuf_free( io->iobuf );
     tr_netClose( io->socket );
     tr_cryptoFree( io->crypto );
     tr_list_free( &io->output_datatypes, tr_free );
+
+    io->magicNumber = 0xDEAD;
     tr_free( io );
 }
@@ -369 +329 @@
 tr_peerIoGetSession( tr_peerIo * io )
 {
-    assert( io );
+    assert( isPeerIo( io ) );
     assert( io->session );
 
@@ -379 +339 @@
                            uint16_t * port )
 {
-    assert( io );
+    assert( isPeerIo( io ) );
 
     if( port )
@@ -407 +367 @@
 tr_peerIoTryRead( tr_peerIo * io )
 {
-    if( EVBUFFER_LENGTH( io->bufev->input ) )
-        canReadWrapper( io->bufev, io );
+    if( EVBUFFER_LENGTH( tr_iobuf_input( io->iobuf )))
+        (*canReadWrapper)( io->iobuf, ~0, io );
 }
 
@@ -444 +404 @@
     if( io->socket >= 0 )
     {
+        tr_bandwidth * bandwidth = io->bandwidth;
+        tr_peerIoSetBandwidth( io, NULL );
+
         tr_netSetTOS( io->socket, io->session->peerSocketTOS );
-
-        bufferevent_free( io->bufev );
+        tr_iobuf_free( io->iobuf );
         bufevNew( io );
+
+        tr_peerIoSetBandwidth( io, bandwidth );
         return 0;
     }
@@ -459 +423 @@
 {
     io->timeout = secs;
-    bufferevent_settimeout( io->bufev, io->timeout, io->timeout );
-    bufferevent_enable( io->bufev, EV_READ | EV_WRITE );
+    tr_iobuf_settimeout( io->iobuf, io->timeout, io->timeout );
+    tr_iobuf_enable( io->iobuf, EV_READ | EV_WRITE );
 }
 
@@ -471 +435 @@
                          const uint8_t * hash )
 {
-    assert( io );
+    assert( isPeerIo( io ) );
 
     tr_cryptoSetTorrentHash( io->crypto, hash );
@@ -479 +443 @@
 tr_peerIoGetTorrentHash( tr_peerIo * io )
 {
-    assert( io );
+    assert( isPeerIo( io ) );
     assert( io->crypto );
 
@@ -488 +452 @@
 tr_peerIoHasTorrentHash( const tr_peerIo * io )
 {
-    assert( io );
+    assert( isPeerIo( io ) );
     assert( io->crypto );
 
@@ -502 +466 @@
                      const uint8_t * peer_id )
 {
-    assert( io );
+    assert( isPeerIo( io ) );
 
     if( ( io->peerIdIsSet = peer_id != NULL ) )
@@ -513 +477 @@
 tr_peerIoGetPeersId( const tr_peerIo * io )
 {
-    assert( io );
+    assert( isPeerIo( io ) );
     assert( io->peerIdIsSet );
 
@@ -527 +491 @@
                      int         flag )
 {
-    assert( io );
+    assert( isPeerIo( io ) );
     assert( flag == 0 || flag == 1 );
 
@@ -537 +501 @@
                      int         flag )
 {
-    assert( io );
+    assert( isPeerIo( io ) );
     assert( flag == 0 || flag == 1 );
 
@@ -546 +510 @@
 tr_peerIoSupportsLTEP( const tr_peerIo * io )
 {
-    assert( io );
+    assert( isPeerIo( io ) );
 
     return io->extendedProtocolSupported;
@@ -554 +518 @@
 tr_peerIoSupportsFEXT( const tr_peerIo * io )
 {
-    assert( io );
+    assert( isPeerIo( io ) );
 
     return io->fastPeersSupported;
@@ -566 +530 @@
 tr_peerIoGetWriteBufferSpace( const tr_peerIo * io )
 {
-    const size_t desiredLiveLen = tr_bandwidthClamp( io->bandwidth[TR_UP], io->session->so_rcvbuf );
-    const size_t currentLiveLen = EVBUFFER_LENGTH( EVBUFFER_OUTPUT( io->bufev ) );
-    const size_t desiredQueueLen = io->session->so_sndbuf;
-    const size_t currentQueueLen = EVBUFFER_LENGTH( io->output );
-    const size_t desiredLen = desiredLiveLen + desiredQueueLen;
-    const size_t currentLen = currentLiveLen + currentQueueLen;
+    const size_t desiredLen = io->session->so_sndbuf * 2; /* FIXME: bigger? */
+    const size_t currentLen = EVBUFFER_LENGTH( tr_iobuf_output( io->iobuf ) );
     size_t freeSpace = 0;
 
@@ -582 +542 @@
 void
 tr_peerIoSetBandwidth( tr_peerIo     * io,
-                       tr_direction    direction,
                        tr_bandwidth  * bandwidth )
 {
-    assert( io );
-    assert( direction == TR_UP || direction == TR_DOWN );
-
-    io->bandwidth[direction] = bandwidth;
-
-    if( direction ==  TR_UP )
-        adjustOutputBuffer( io );
-    else
-        adjustInputBuffer( io );
+    assert( isPeerIo( io ) );
+
+    if( io->bandwidth )
+        tr_bandwidthRemoveBuffer( io->bandwidth, io->iobuf );
+
+    io->bandwidth = bandwidth;
+    tr_iobuf_set_bandwidth( io->iobuf, bandwidth );
+
+    if( io->bandwidth )
+        tr_bandwidthAddBuffer( io->bandwidth, io->iobuf );
+
+    tr_iobuf_enable( io->iobuf, EV_READ | EV_WRITE );
 }
 
@@ -610 +572 @@
                         int         encryptionMode )
 {
-    assert( io );
+    assert( isPeerIo( io ) );
     assert( encryptionMode == PEER_ENCRYPTION_NONE
           || encryptionMode == PEER_ENCRYPTION_RC4 );
@@ -637 +599 @@
     dbgmsg( io, "adding %zu bytes into io->output", writemeLen );
 
-    evbuffer_add( io->output, writeme, writemeLen );
-
     datatype = tr_new( struct tr_datatype, 1 );
     datatype->isPieceData = isPieceData != 0;
@@ -644 +604 @@
     tr_list_append( &io->output_datatypes, datatype );
 
-    adjustOutputBuffer( io );
+    evbuffer_add( tr_iobuf_output( io->iobuf ), writeme, writemeLen );
+    tr_iobuf_enable( io->iobuf, EV_WRITE );
 }
 

trunk/libtransmission/peer-io.h

@@ -24 +24 @@
 struct in_addr;
 struct evbuffer;
-struct bufferevent;
 struct tr_bandwidth;
 struct tr_crypto;
+struct tr_iobuf;
 typedef struct tr_peerIo tr_peerIo;
 
@@ -111 +111 @@
 ReadState;
 
-typedef ReadState ( *tr_can_read_cb  )( struct bufferevent   * ev,
-                                        void                 * user_data );
-
-typedef void      ( *tr_did_write_cb )( tr_peerIo            * io,
-                                        size_t                 bytesWritten,
-                                        int                    wasPieceData,
-                                        void                 * userData );
-
-typedef void      ( *tr_net_error_cb )( struct bufferevent  * ev,
-                                        short                 what,
-                                        void                * userData );
+typedef ReadState ( *tr_can_read_cb  )( struct tr_iobuf  * iobuf,
+                                        void             * user_data,
+                                        size_t           * setme_piece_byte_count );
+
+typedef void      ( *tr_did_write_cb )( tr_peerIo        * io,
+                                        size_t             bytesWritten,
+                                        int                wasPieceData,
+                                        void             * userData );
+
+typedef void      ( *tr_net_error_cb )( struct tr_iobuf  * ev,
+                                        short              what,
+                                        void             * userData );
 
 void    tr_peerIoSetIOFuncs      ( tr_peerIo        * io,
@@ -206 +207 @@
 
 void              tr_peerIoSetBandwidth( tr_peerIo            * io,
-                                         tr_direction           direction,
                                          struct tr_bandwidth  * bandwidth );
 
+void              tr_peerIoBandwidthUsed( tr_peerIo           * io,
+                                          tr_direction          direction,
+                                          size_t                byteCount,
+                                          int                   isPieceData );
+
+
 
 #endif

trunk/libtransmission/peer-mgr-private.h

@@ -28 +28 @@
 #include "publish.h" /* tr_publisher_tag */
 
+struct tr_bandwidth;
 struct tr_bitfield;
 struct tr_peerIo;
 struct tr_peermsgs;
-struct tr_ratecontrol;
 
 enum
@@ -72 +72 @@
     tr_publisher_tag         msgsTag;
 
-    /* the rate at which pieces are being transferred between client and peer.
-     * protocol overhead is NOT included; this is only the piece data */
-    struct tr_ratecontrol  * pieceSpeed[2];
-
-    /* the rate at which all data is being transferred between client and peer. */
-    struct tr_ratecontrol  * rawSpeed[2];
+    struct tr_bandwidth    * bandwidth;
 }
 tr_peer;

trunk/libtransmission/peer-mgr.c

@@ -34 +34 @@
 #include "peer-msgs.h"
 #include "ptrarray.h"
-#include "ratecontrol.h"
 #include "stats.h" /* tr_statsAddUploaded, tr_statsAddDownloaded */
 #include "torrent.h"
@@ -59 +58 @@
    
     /* how frequently to reallocate bandwidth */
-    BANDWIDTH_PERIOD_MSEC = 333,
+    BANDWIDTH_PERIOD_MSEC = 200,
 
     /* max # of peers to ask fer per torrent per reconnect pulse */
@@ -324 +323 @@
 
 static tr_peer*
-peerConstructor( const struct in_addr * in_addr )
+peerConstructor( tr_torrent * tor, const struct in_addr * in_addr )
 {
     tr_peer * p;
@@ -330 +329 @@
     p = tr_new0( tr_peer, 1 );
     memcpy( &p->in_addr, in_addr, sizeof( struct in_addr ) );
-    p->pieceSpeed[TR_CLIENT_TO_PEER] = tr_rcInit( );
-    p->pieceSpeed[TR_PEER_TO_CLIENT] = tr_rcInit( );
-    p->rawSpeed[TR_CLIENT_TO_PEER] = tr_rcInit( );
-    p->rawSpeed[TR_PEER_TO_CLIENT] = tr_rcInit( );
+    p->bandwidth = tr_bandwidthNew( tor->session, tor->bandwidth );
     return p;
 }
@@ -349 +345 @@
     if( peer == NULL )
     {
-        peer = peerConstructor( in_addr );
+        peer = peerConstructor( torrent->tor, in_addr );
         tr_ptrArrayInsertSorted( torrent->peers, peer, peerCompare );
     }
@@ -371 +367 @@
     tr_free( peer->client );
 
-    tr_rcClose( peer->rawSpeed[TR_CLIENT_TO_PEER] );
-    tr_rcClose( peer->rawSpeed[TR_PEER_TO_CLIENT] );
-    tr_rcClose( peer->pieceSpeed[TR_CLIENT_TO_PEER] );
-    tr_rcClose( peer->pieceSpeed[TR_PEER_TO_CLIENT] );
+    tr_bandwidthFree( peer->bandwidth );
+
     tr_free( peer );
 }
@@ -1018 +1012 @@
             const time_t now = time( NULL );
             tr_torrent * tor = t->tor;
-            const tr_direction dir = TR_CLIENT_TO_PEER;
 
             tor->activityDate = now;
@@ -1024 +1017 @@
             if( e->wasPieceData )
                 tor->uploadedCur += e->length;
-
-            /* add it to the raw upload speed */
-            if( peer )
-                tr_rcTransferred ( peer->rawSpeed[dir], e->length );
-
-            /* maybe add it to the piece upload speed */
-            if( e->wasPieceData ) {
-                if( peer )
-                    tr_rcTransferred ( peer->pieceSpeed[dir], e->length );
-            }
-
-            tr_bandwidthUsed( tor->bandwidth[dir], e->length, e->wasPieceData );
-            tr_bandwidthUsed( tor->session->bandwidth[dir], e->length, e->wasPieceData );
 
             /* update the stats */
@@ -1055 +1035 @@
             const time_t now = time( NULL );
             tr_torrent * tor = t->tor;
-            const tr_direction dir = TR_PEER_TO_CLIENT;
 
             tor->activityDate = now;
@@ -1068 +1047 @@
                 tor->downloadedCur += e->length;
 
-            /* add it to our raw download speed */
-            if( peer )
-                tr_rcTransferred ( peer->rawSpeed[dir], e->length );
-
-            /* maybe add it to the piece upload speed */
-            if( e->wasPieceData ) {
-                if( peer )
-                    tr_rcTransferred ( peer->pieceSpeed[dir], e->length );
-            }
-
-            tr_bandwidthUsed( tor->bandwidth[dir], e->length, e->wasPieceData );
-            tr_bandwidthUsed( tor->session->bandwidth[dir], e->length, e->wasPieceData );
-     
             /* update the stats */ 
             if( e->wasPieceData )
@@ -1313 +1279 @@
                 if( !peer_id )
                     peer->client = NULL;
-                else
-                {
+                else {
                     char client[128];
                     tr_clientForId( client, sizeof( client ), peer_id );
                     peer->client = tr_strdup( client );
                 }
+
                 peer->port = port;
                 peer->io = io;
-                peer->msgs =
-                    tr_peerMsgsNew( t->tor, peer, peerCallbackFunc, t,
-                                    &peer->msgsTag );
+                peer->msgs = tr_peerMsgsNew( t->tor, peer, peerCallbackFunc, t, &peer->msgsTag );
+                tr_peerIoSetBandwidth( io, peer->bandwidth );
 
                 success = TRUE;
@@ -1805 +1770 @@
     assert( direction==TR_CLIENT_TO_PEER || direction==TR_PEER_TO_CLIENT );
 
-    return tr_rcRate( peer->pieceSpeed[direction] );
+    return tr_bandwidthGetPieceSpeed( peer->bandwidth, direction );
 }
 
@@ -2363 +2328 @@
     int       i, j;
 
-    for( i = 0; i < torrentCount; ++i )
+    for( i=0; i<torrentCount; ++i )
     {
         Torrent * t = tr_ptrArrayNth( mgr->torrents, i );
-        for( j = 0; j < tr_ptrArraySize( t->peers ); ++j )
+        for( j=0; j<tr_ptrArraySize( t->peers ); ++j )
         {
             tr_peer * peer = tr_ptrArrayNth( t->peers, j );
@@ -2374 +2339 @@
 }
 
-static void
-setTorrentBandwidth( Torrent * t,
-                     tr_direction dir,
-                     tr_bandwidth * bandwidth )
-{
-    int i;
-    int peerCount = 0;
-    tr_peer ** peers = getConnectedPeers( t, &peerCount );
-
-    for( i=0; i<peerCount; ++i )
-        tr_peerIoSetBandwidth( peers[i]->io, dir, bandwidth );
-
-    tr_free( peers );
-}
-
-#if 0
-#define DEBUG_DIRECTION TR_DOWN
-#endif
-
-static double
-bytesPerPulse( double KiB_per_second )
-{
-    return KiB_per_second * ( 1024.0 * BANDWIDTH_PERIOD_MSEC / 1000.0 );
-}
-
-/*
- * @param currentSpeed current speed in KiB/s
- * @param desiredSpeed desired speed in KiB/s
- */
-static double
-allocateHowMuch( tr_direction dir UNUSED, double currentSpeed, double desiredSpeed )
-{
-    const int pulses_per_history = TR_RATECONTROL_HISTORY_MSEC / BANDWIDTH_PERIOD_MSEC;
-    const double current_bytes_per_pulse = bytesPerPulse( currentSpeed );
-    const double desired_bytes_per_pulse = bytesPerPulse( desiredSpeed );
-    const double min = desired_bytes_per_pulse * 0.90;
-    const double max = desired_bytes_per_pulse * 1.50;
-    const double next_pulse_bytes = desired_bytes_per_pulse * ( pulses_per_history + 1 )
-                                  - ( current_bytes_per_pulse * pulses_per_history );
-    double clamped;
-
-    /* clamp the return value to lessen oscillation */
-    clamped = next_pulse_bytes;
-    clamped = MAX( clamped, min );
-    clamped = MIN( clamped, max );
-
-#ifdef DEBUG_DIRECTION
-if( dir == DEBUG_DIRECTION )
-fprintf( stderr, "currentSpeed(%5.2f) desiredSpeed(%5.2f), allocating %5.2f (unclamped: %5.2f)\n",
-         currentSpeed,
-         desiredSpeed,
-         clamped/1024.0,
-         next_pulse_bytes/1024.0 );
-#endif
-
-    return clamped;
-}
-
-/**
- * Allocate bandwidth for each peer connection.
- *
- * @param mgr the peer manager
- * @param direction whether to allocate upload or download bandwidth
- */
-static void
-allocateBandwidth( tr_peerMgr * mgr,
-                   tr_direction direction )
-{
-    int              i;
-    tr_session     * session = mgr->session;
-    const int        torrentCount = tr_ptrArraySize( mgr->torrents );
-    Torrent       ** torrents = (Torrent **) tr_ptrArrayBase( mgr->torrents );
-    tr_bandwidth   * global_pool = session->bandwidth[direction];
-
-    assert( mgr );
-    assert( direction == TR_UP || direction == TR_DOWN );
-
-    /* before allocating bandwidth, pump the connected peers */
+static int
+bandwidthPulse( void * vmgr )
+{
+    tr_peerMgr * mgr = vmgr;
+    managerLock( mgr );
+
     pumpAllPeers( mgr );
-
-    for( i=0; i<torrentCount; ++i )
-    {
-        Torrent * t = torrents[i];
-        tr_speedlimit speedMode;
-
-        /* no point in allocating bandwidth for stopped torrents */
-        if( tr_torrentGetActivity( t->tor ) == TR_STATUS_STOPPED )
-            continue;
-
-        /* if piece data is disallowed, don't bother limiting bandwidth --
-         * we won't be asking for, or sending out, any pieces */
-        if( !tr_torrentIsPieceTransferAllowed( t->tor, direction ) )
-            speedMode = TR_SPEEDLIMIT_UNLIMITED;
-        else
-            speedMode = tr_torrentGetSpeedMode( t->tor, direction );
-            
-        /* process the torrent's peers based on its speed mode */
-        switch( speedMode )
-        {
-            case TR_SPEEDLIMIT_UNLIMITED:
-            {
-                tr_bandwidth * b = t->tor->bandwidth[direction];
-                tr_bandwidthSetUnlimited( b );
-                setTorrentBandwidth( t, direction, b );
-                break;
-            }
-
-            case TR_SPEEDLIMIT_SINGLE:
-            {
-                tr_bandwidth * b = t->tor->bandwidth[direction];
-                const double currentSpeed = tr_bandwidthGetPieceSpeed( b );
-                const double desiredSpeed = tr_torrentGetSpeedLimit( t->tor, direction );
-                const double bytesPerPulse = allocateHowMuch( direction, currentSpeed, desiredSpeed );
-#ifdef DEBUG_DIRECTION
-if( direction == DEBUG_DIRECTION )
-fprintf( stderr, "single: currentSpeed %.0f ... desiredSpeed %.0f ... bytesPerPulse %.0f\n", currentSpeed, desiredSpeed, bytesPerPulse );
-#endif
-                tr_bandwidthSetLimited( b, bytesPerPulse );
-                setTorrentBandwidth( t, direction, b );
-                break;
-            }
-
-            case TR_SPEEDLIMIT_GLOBAL:
-            {
-                setTorrentBandwidth( t, direction, global_pool );
-                break;
-            }
-        }
-    }
-
-    /* handle the global pool's connections */
-    if( !tr_sessionIsSpeedLimitEnabled( session, direction ) )
-        tr_bandwidthSetUnlimited( global_pool );
-    else {
-        const double currentSpeed = tr_bandwidthGetPieceSpeed( global_pool );
-        const double desiredSpeed = tr_sessionGetSpeedLimit( session, direction );
-        const double bytesPerPulse = allocateHowMuch( direction, currentSpeed, desiredSpeed );
-#ifdef DEBUG_DIRECTION
-if( direction == DEBUG_DIRECTION )
-fprintf( stderr, "global(%p): currentSpeed %.0f ... desiredSpeed %.0f ... bytesPerPulse %.0f\n", global_pool, currentSpeed, desiredSpeed, bytesPerPulse );
-#endif
-        tr_bandwidthSetLimited( session->bandwidth[direction], bytesPerPulse );
-    }
-
-    /* now that we've allocated bandwidth, pump all the connected peers */
+    tr_bandwidthAllocate( mgr->session->bandwidth, TR_UP, BANDWIDTH_PERIOD_MSEC );
+    tr_bandwidthAllocate( mgr->session->bandwidth, TR_DOWN, BANDWIDTH_PERIOD_MSEC );
     pumpAllPeers( mgr );
-}
-
-static int
-bandwidthPulse( void * vmgr )
-{
-    tr_peerMgr * mgr = vmgr;
-    int          i;
-
-    managerLock( mgr );
-
-    /* allocate the upload and download bandwidth */
-    for( i = 0; i < 2; ++i )
-        allocateBandwidth( mgr, i );
 
     managerUnlock( mgr );

trunk/libtransmission/peer-msgs.c

@@ -25 +25 @@
 #include "crypto.h"
 #include "inout.h"
+#include "iobuf.h"
 #ifdef WIN32
 #include "net.h" /* for ECONN */
@@ -1315 +1316 @@
 
 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;
@@ -1350 +1352 @@
         evbuffer_add( msgs->incoming.block, buf, 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",
@@ -1643 +1646 @@
 
 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 );
 
@@ -1657 +1659 @@
     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 )
@@ -1821 +1823 @@
 
 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)",

trunk/libtransmission/session.c

@@ -275 +275 @@
     /* Initialize rate and file descripts controls */
 
-    h->uploadLimit = uploadLimit;
-    h->useUploadLimit = useUploadLimit;
-    h->downloadLimit = downloadLimit;
-    h->useDownloadLimit = useDownloadLimit;
-
     tr_fdInit( globalPeerLimit );
     h->shared = tr_sharedInit( h, isPortForwardingEnabled, publicPort );
     h->isPortSet = publicPort >= 0;
 
-    h->bandwidth[TR_UP] = tr_bandwidthNew( h );
-    h->bandwidth[TR_DOWN] = tr_bandwidthNew( h );
+    h->bandwidth = tr_bandwidthNew( h, NULL );
+    tr_bandwidthSetDesiredSpeed( h->bandwidth, TR_UP, uploadLimit );
+    tr_bandwidthSetDesiredSpeed( h->bandwidth, TR_DOWN, downloadLimit );
+    tr_bandwidthSetLimited( h->bandwidth, TR_UP, useUploadLimit );
+    tr_bandwidthSetLimited( h->bandwidth, TR_DOWN, useDownloadLimit );
 
     /* first %s is the application name
@@ -444 +442 @@
 
 void
-tr_sessionSetSpeedLimitEnabled( tr_handle *  h,
-                                tr_direction direction,
-                                int          use_flag )
-{
-    assert( h );
-    assert( direction == TR_UP || direction == TR_DOWN );
-
-    if( direction == TR_UP )
-        h->useUploadLimit = use_flag ? 1 : 0;
-    else
-        h->useDownloadLimit = use_flag ? 1 : 0;
-}
-
-int
-tr_sessionIsSpeedLimitEnabled( const tr_handle * h,
-                               tr_direction      direction )
-{
-    return direction == TR_UP ? h->useUploadLimit : h->useDownloadLimit;
-}
-
-void
-tr_sessionSetSpeedLimit( tr_handle *  h,
-                         tr_direction direction,
-                         int          KiB_sec )
-{
-    if( direction == TR_DOWN )
-        h->downloadLimit = KiB_sec;
-    else
-        h->uploadLimit = KiB_sec;
-}
-
-int
-tr_sessionGetSpeedLimit( const tr_handle * h,
-                         tr_direction      direction )
-{
-    return direction == TR_UP ? h->uploadLimit : h->downloadLimit;
+tr_sessionSetSpeedLimitEnabled( tr_session      * session,
+                                tr_direction      dir,
+                                int               isLimited )
+{
+    tr_bandwidthSetLimited( session->bandwidth, dir, isLimited );
+}
+
+int
+tr_sessionIsSpeedLimitEnabled( const tr_session  * session,
+                               tr_direction        dir )
+{
+    return !tr_bandwidthIsLimited( session->bandwidth, dir );
+}
+
+void
+tr_sessionSetSpeedLimit( tr_session    * session,
+                         tr_direction    dir,
+                         int             desiredSpeed )
+{
+    tr_bandwidthSetDesiredSpeed( session->bandwidth, dir, desiredSpeed );
+}
+
+int
+tr_sessionGetSpeedLimit( const tr_session  * session,
+                         tr_direction        dir )
+{
+    return tr_bandwidthGetDesiredSpeed( session->bandwidth, dir );
 }
 
@@ -506 +495 @@
 tr_sessionGetPieceSpeed( const tr_session * session, tr_direction dir )
 {
-    assert( dir==TR_UP || dir==TR_DOWN );
-
-    return session ? tr_bandwidthGetPieceSpeed( session->bandwidth[dir] ) : 0.0;
+    return session ? tr_bandwidthGetPieceSpeed( session->bandwidth, dir ) : 0.0;
 }
 
@@ -514 +501 @@
 tr_sessionGetRawSpeed( const tr_session * session, tr_direction dir )
 {
-    assert( dir==TR_UP || dir==TR_DOWN );
-
-    return session ? tr_bandwidthGetPieceSpeed( session->bandwidth[dir] ) : 0.0;
+    return session ? tr_bandwidthGetPieceSpeed( session->bandwidth, dir ) : 0.0;
 }
 
@@ -630 +615 @@
 
     /* free the session memory */
-    tr_bandwidthFree( session->bandwidth[TR_UP] );
-    tr_bandwidthFree( session->bandwidth[TR_DOWN] );
+    tr_bandwidthFree( session->bandwidth );
     tr_lockFree( session->lock );
     for( i = 0; i < session->metainfoLookupCount; ++i )

trunk/libtransmission/session.h

@@ -63 +63 @@
     unsigned int                 isProxyAuthEnabled : 1;
     unsigned int                 isClosed           : 1;
-    unsigned int                 useUploadLimit     : 1;
-    unsigned int                 useDownloadLimit   : 1;
     unsigned int                 useLazyBitfield    : 1;
 
@@ -88 +86 @@
     char *                       proxyUsername;
     char *                       proxyPassword;
-
-    int                          uploadLimit;
-    int                          downloadLimit;
 
     struct tr_list *             blocklists;
@@ -117 +112 @@
 
     /* monitors the "global pool" speeds */
-    struct tr_bandwidth       * bandwidth[2];
+    struct tr_bandwidth       * bandwidth;
 };
 

trunk/libtransmission/torrent.c

@@ -144 +144 @@
 void
 tr_torrentSetSpeedMode( tr_torrent *  tor,
-                        tr_direction  direction,
+                        tr_direction  dir,
                         tr_speedlimit mode )
 {
-    tr_speedlimit * limit = direction == TR_UP ? &tor->uploadLimitMode
-                                               : &tor->downloadLimitMode;
-
-    *limit = mode;
+    assert( tor != NULL );
+    assert( dir==TR_UP || dir==TR_DOWN );
+    assert( mode==TR_SPEEDLIMIT_GLOBAL || mode==TR_SPEEDLIMIT_SINGLE || mode==TR_SPEEDLIMIT_UNLIMITED  );
+
+    tor->speedLimitMode[dir] = mode;
+
+    tr_bandwidthSetLimited( tor->bandwidth, dir, mode==TR_SPEEDLIMIT_SINGLE );
+    tr_bandwidthHonorParentLimits( tor->bandwidth, dir, mode!=TR_SPEEDLIMIT_UNLIMITED );
 }
 
 tr_speedlimit
 tr_torrentGetSpeedMode( const tr_torrent * tor,
-                        tr_direction       direction )
-{
-    return direction == TR_UP ? tor->uploadLimitMode
-                              : tor->downloadLimitMode;
+                        tr_direction       dir )
+{
+    assert( tor != NULL );
+    assert( dir==TR_UP || dir==TR_DOWN );
+
+    return tor->speedLimitMode[dir];
 }
 
 void
 tr_torrentSetSpeedLimit( tr_torrent * tor,
-                         tr_direction direction,
-                         int          single_KiB_sec )
-{
-    switch( direction )
-    {
-        case TR_UP:
-            tor->uploadLimit = single_KiB_sec; break;
-
-        case TR_DOWN:
-            tor->downloadLimit = single_KiB_sec; break;
-
-        default:
-            assert( 0 );
-    }
+                         tr_direction dir,
+                         int          desiredSpeed )
+{
+    tr_bandwidthSetDesiredSpeed( tor->bandwidth, dir, desiredSpeed );
 }
 
 int
 tr_torrentGetSpeedLimit( const tr_torrent * tor,
-                         tr_direction       direction )
-{
-    switch( direction )
-    {
-        case TR_UP:
-            return tor->uploadLimit;
-
-        case TR_DOWN:
-            return tor->downloadLimit;
-
-        default:
-            assert( 0 );
-    }
+                         tr_direction       dir )
+{
+    return tr_bandwidthGetDesiredSpeed( tor->bandwidth, dir );
 }
 
@@ -498 +484 @@
     randomizeTiers( info );
 
-    tor->bandwidth[TR_UP] = tr_bandwidthNew( h );
-    tor->bandwidth[TR_DOWN] = tr_bandwidthNew( h );
+    tor->bandwidth = tr_bandwidthNew( h, h->bandwidth );
+
+fprintf( stderr, "torrent [%s] bandwidth is %p\n", info->name, tor->bandwidth );
 
     tor->blockSize = getBlockSize( info->pieceSize );
@@ -541 +528 @@
     tr_torrentInitFilePieces( tor );
 
-    tor->uploadLimit = 0;
-    tor->downloadLimit = 0;
     tor->swarmSpeed = tr_rcInit( );
 
@@ -814 +799 @@
                             s->peersFrom );
 
-    s->rawUploadSpeed     = tr_bandwidthGetRawSpeed  ( tor->bandwidth[TR_UP] );
-    s->rawDownloadSpeed   = tr_bandwidthGetRawSpeed  ( tor->bandwidth[TR_DOWN] );
-    s->pieceUploadSpeed   = tr_bandwidthGetPieceSpeed( tor->bandwidth[TR_UP] );
-    s->pieceDownloadSpeed = tr_bandwidthGetPieceSpeed( tor->bandwidth[TR_DOWN] );
+    s->rawUploadSpeed     = tr_bandwidthGetRawSpeed  ( tor->bandwidth, TR_UP );
+    s->rawDownloadSpeed   = tr_bandwidthGetRawSpeed  ( tor->bandwidth, TR_DOWN );
+    s->pieceUploadSpeed   = tr_bandwidthGetPieceSpeed( tor->bandwidth, TR_UP );
+    s->pieceDownloadSpeed = tr_bandwidthGetPieceSpeed( tor->bandwidth, TR_DOWN );
 
     usableSeeds += tor->info.webseedCount;
@@ -1100 +1085 @@
     h->torrentCount--;
 
-    tr_bandwidthFree( tor->bandwidth[TR_DOWN] );
-    tr_bandwidthFree( tor->bandwidth[TR_UP] );
+    tr_bandwidthFree( tor->bandwidth );
 
     tr_metainfoFree( inf );

trunk/libtransmission/torrent.h

@@ -173 +173 @@
     tr_info                  info;
 
-    int                      uploadLimit;
-    tr_speedlimit            uploadLimitMode;
-    int                      downloadLimit;
-    tr_speedlimit            downloadLimitMode;
+    tr_speedlimit            speedLimitMode[2];
 
     struct tr_ratecontrol *  swarmSpeed;
@@ -235 +232 @@
     int                        uniqueId;
 
-    struct tr_bandwidth      * bandwidth[2];
+    struct tr_bandwidth      * bandwidth;
 };