Changeset 7455

Timestamp:

Dec 22, 2008, 12:51:14 AM (12 years ago)

Author:

charles

Message:

(1.4x libT) backport handshake, peer, bandwidth, peer-io to 1.4x.

Location:

branches/1.4x

Files:

• configure.ac (modified) (1 diff)
• libtransmission/Makefile.am (modified) (2 diffs)
• libtransmission/bandwidth.c (modified) (20 diffs)
• libtransmission/bandwidth.h (modified) (7 diffs)
• libtransmission/bencode-test.c (modified) (1 diff)
• libtransmission/fastresume.c (modified) (1 diff)
• libtransmission/handshake.c (modified) (31 diffs)
• libtransmission/handshake.h (modified) (4 diffs)
• libtransmission/net.c (modified) (2 diffs)
• libtransmission/net.h (modified) (2 diffs)
• libtransmission/peer-common.h (modified) (3 diffs)
• libtransmission/peer-io.c (modified) (35 diffs)
• libtransmission/peer-io.h (modified) (7 diffs)
• libtransmission/peer-mgr-private.h (modified) (1 diff)
• libtransmission/peer-mgr.c (modified) (88 diffs)
• libtransmission/peer-mgr.h (modified) (2 diffs)
• libtransmission/peer-msgs-test.c (added)
• libtransmission/peer-msgs.c (modified) (90 diffs)
• libtransmission/peer-msgs.h (modified) (1 diff)
• libtransmission/resume.c (modified) (1 diff)
• libtransmission/test-fastset.c (deleted)
• libtransmission/transmission.h (modified) (3 diffs)
• libtransmission/utils.c (modified) (1 diff)
• libtransmission/webseed.c (modified) (7 diffs)
• libtransmission/webseed.h (modified) (1 diff)
• macosx/version.sh (modified) (1 diff)

branches/1.4x/configure.ac

@@ -4 +4 @@
 dnl         "0" for stable, supported releases
 dnl these should be the only two lines you need to change
-m4_define([user_agent_prefix],1.41)
-m4_define([peer_id_prefix],-TR1410-)
+m4_define([user_agent_prefix],[1.41b4])
+m4_define([peer_id_prefix],[-TR141X-])
 
 AC_INIT([transmission],[user_agent_prefix],[http://trac.transmissionbt.com/newticket])

branches/1.4x/libtransmission/Makefile.am

@@ -110 +110 @@
     clients-test \
     json-test \
+    peer-msgs-test \
     rpc-test \
-    test-fastset \
     test-peer-id \
     utils-test
@@ -152 +152 @@
 rpc_test_LDFLAGS = ${apps_ldflags}
 
-test_fastset_SOURCES = test-fastset.c
-test_fastset_LDADD = ${apps_ldadd}
-test_fastset_LDFLAGS = ${apps_ldflags}
+peer_msgs_test_SOURCES = peer-msgs-test.c
+peer_msgs_test_LDADD = ${apps_ldadd}
+peer_msgs_test_LDFLAGS = ${apps_ldflags}
 
 test_peer_id_SOURCES = test-peer-id.c

branches/1.4x/libtransmission/bandwidth.c

@@ -1 +1 @@
 /*
- * This file Copyright (C) 2008 Charles Kerr <charles@rebelbase.com>
+ * This file Copyright (C) 2008 Charles Kerr <charles@transmissionbt.com>
  *
  * This file is licensed by the GPL version 2.  Works owned by the
@@ -19 +19 @@
 #include "bandwidth.h"
 #include "crypto.h"
-#include "iobuf.h"
+#include "peer-io.h"
 #include "ptrarray.h"
 #include "utils.h"
@@ -100 +100 @@
     tr_session * session;
     tr_ptrArray * children; /* struct tr_bandwidth */
-    tr_ptrArray * iobufs; /* struct tr_iobuf */
+    tr_ptrArray * peers; /* tr_peerIo */
 };
 
@@ -113 +113 @@
 }
 
-static int
-isBandwidth( const tr_bandwidth * b )
+tr_bool
+tr_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 );
 }
 
@@ -135 +129 @@
     b->session = session;
     b->children = tr_ptrArrayNew( );
-    b->iobufs = tr_ptrArrayNew( );
+    b->peers = tr_ptrArrayNew( );
     b->magicNumber = MAGIC_NUMBER;
-    b->band[TR_UP].honorParentLimits = 1;
-    b->band[TR_DOWN].honorParentLimits = 1;
+    b->band[TR_UP].honorParentLimits = TRUE;
+    b->band[TR_DOWN].honorParentLimits = TRUE;
     tr_bandwidthSetParent( b, parent );
     return b;
@@ -146 +140 @@
 tr_bandwidthFree( tr_bandwidth * b )
 {
-    assert( isBandwidth( b ) );
+    assert( tr_isBandwidth( b ) );
 
     tr_bandwidthSetParent( b, NULL );
-    tr_ptrArrayFree( b->iobufs, NULL );
+    tr_ptrArrayFree( b->peers, NULL );
     tr_ptrArrayFree( b->children, NULL );
     b->magicNumber = 0xDEAD;
@@ -163 +157 @@
                        tr_bandwidth  * parent )
 {
-    assert( isBandwidth( b ) );
+    assert( tr_isBandwidth( b ) );
     assert( b != parent );
 
     if( b->parent )
     {
-        assert( isBandwidth( b->parent ) );
+        assert( tr_isBandwidth( b->parent ) );
 
         tr_ptrArrayRemoveSorted( b->parent->children, b, comparePointers );
-        b->parent= NULL;
+        b->parent = NULL;
     }
 
     if( parent )
     {
-        assert( isBandwidth( parent ) );
+        assert( tr_isBandwidth( parent ) );
         assert( parent->parent != b );
 
@@ -187 +181 @@
 tr_bandwidthHonorParentLimits( tr_bandwidth  * b,
                                tr_direction    dir,
-                               int             honorParentLimits )
-{
-    assert( isBandwidth( b ) );
-    assert( isDirection( dir ) );
-
-    b->band[dir].honorParentLimits = honorParentLimits != 0;
+                               tr_bool         honorParentLimits )
+{
+    assert( tr_isBandwidth( b ) );
+    assert( tr_isDirection( dir ) );
+
+    b->band[dir].honorParentLimits = honorParentLimits;
 }
 
@@ -204 +198 @@
                              double          desiredSpeed )
 {
-    assert( isBandwidth( b ) );
-    assert( isDirection( dir ) );
+    assert( tr_isBandwidth( b ) );
+    assert( tr_isDirection( dir ) );
 
     b->band[dir].desiredSpeed = desiredSpeed; 
@@ -214 +208 @@
                              tr_direction          dir )
 {
-    assert( isBandwidth( b ) );
-    assert( isDirection( dir ) );
+    assert( tr_isBandwidth( b ) );
+    assert( tr_isDirection( dir ) );
 
     return b->band[dir].desiredSpeed;
@@ -223 +217 @@
 tr_bandwidthSetLimited( tr_bandwidth  * b,
                         tr_direction    dir,
-                        int             isLimited )
-{
-    assert( isBandwidth( b ) );
-    assert( isDirection( dir ) );
-
-    b->band[dir].isLimited = isLimited != 0;
-}
-
-int
+                        tr_bool         isLimited )
+{
+    assert( tr_isBandwidth( b ) );
+    assert( tr_isDirection( dir ) );
+
+    b->band[dir].isLimited = isLimited;
+}
+
+tr_bool
 tr_bandwidthIsLimited( const tr_bandwidth  * b,
                        tr_direction          dir )
 {
-    assert( isBandwidth( b ) );
-    assert( isDirection( dir ) );
-
-    return b->band[dir].isLimited != 0;
+    assert( tr_isBandwidth( b ) );
+    assert( tr_isDirection( dir ) );
+
+    return b->band[dir].isLimited;
 }
 
@@ -250 +244 @@
                    tr_direction    dir,
                    int             period_msec,
-                   tr_ptrArray   * iobuf_pool )
-{
-    assert( isBandwidth( b ) );
-    assert( isDirection( dir ) );
-
+                   tr_ptrArray   * peer_pool )
+{
+    assert( tr_isBandwidth( b ) );
+    assert( tr_isDirection( dir ) );
+
+    /* set the available bandwidth */
     if( b->band[dir].isLimited )
     {
@@ -269 +264 @@
     }
 
+    /* traverse & repeat for the subtree */
     {
         int i;
-        const int n = tr_ptrArraySize( b->iobufs );
+        const int n = tr_ptrArraySize( b->peers );
         for( i=0; i<n; ++i )
-            tr_ptrArrayAppend( iobuf_pool, tr_ptrArrayNth( b->iobufs, i ) );
+            tr_ptrArrayAppend( peer_pool, tr_ptrArrayNth( b->peers, i ) );
     }
 
 #ifdef DEBUG_DIRECTION
 if( ( dir == DEBUG_DIRECTION ) && ( n > 1 ) )
-fprintf( stderr, "bandwidth %p has %d iobufs\n", b, n );
+fprintf( stderr, "bandwidth %p has %d peers\n", b, n );
 #endif
 
@@ -286 +282 @@
         struct tr_bandwidth ** children = (struct tr_bandwidth**) tr_ptrArrayPeek( b->children, &n );
         for( i=0; i<n; ++i )
-            allocateBandwidth( children[i], dir, period_msec, iobuf_pool );
+            allocateBandwidth( children[i], dir, period_msec, peer_pool );
     }
 }
@@ -295 +291 @@
                       int             period_msec )
 {
-    int n;
+    int i, n, peerCount;
     tr_ptrArray * tmp;
-    struct tr_iobuf ** buffers;
-    const size_t chunkSize = 1024; /* arbitrary */
-
+    struct tr_peerIo ** peers;
+
+    /* allocateBandwidth() is a helper function with two purposes:
+     * 1. allocate bandwidth to b and its subtree
+     * 2. accumulate an array of all the peerIos from b and its subtree. */
     tmp = tr_ptrArrayNew( );
     allocateBandwidth( b, dir, period_msec, tmp );
-    buffers = (struct tr_iobuf**) tr_ptrArrayPeek( tmp, &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];
+    peers = (struct tr_peerIo**) tr_ptrArrayPeek( tmp, &peerCount );
+
+    /* Stop all peers from listening for the socket to be ready for IO.
+     * See "Second phase of IO" lower in this function for more info. */
+    for( i=0; i<peerCount; ++i )
+        tr_peerIoSetEnabled( peers[i], dir, FALSE );
+
+    /* First phase of IO.  Tries to distribute bandwidth fairly to keep faster
+     * peers from starving the others.  Loop through the peers, giving each a
+     * small chunk of bandwidth.  Keep looping until we run out of bandwidth
+     * or pweers that can use it */
+    n = peerCount;
+    i = n ? tr_cryptoWeakRandInt( n ) : 0; /* pick a random starting point */
+    for( ; n>0; )
+    {
+        const int increment = n==1 ? 4096 : 1024;
+        const int byteCount = tr_peerIoFlush( peers[i], dir, increment);
+
+        if( byteCount == increment )
+            ++i;
+        else {
+            /* peer is done writing for now; move it to the end of the list */
+            tr_peerIo * tmp = peers[i];
+            peers[i] = peers[n-1];
+            peers[n-1] = tmp;
+            --n;
         }
-    }
+
+        assert( i <= n );
+        if( i == n )
+            i = 0;
+    }
+
+    /* Second phase of IO.  To help us scale in high bandwidth situations,
+     * enable on-demand IO for peers with bandwidth left to burn.
+     * This on-demand IO is enabled until (1) the peer runs out of bandwidth,
+     * or (2) the next tr_bandwidthAllocate() call, when we start over again. */
+    for( i=0; i<peerCount; ++i )
+        if( tr_peerIoHasBandwidthLeft( peers[i], dir ) )
+            tr_peerIoSetEnabled( peers[i], dir, TRUE );
 
     /* cleanup */
@@ -333 +350 @@
 
 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 );
+tr_bandwidthAddPeer( tr_bandwidth   * b,
+                     tr_peerIo      * peerIo )
+{
+    assert( tr_isBandwidth( b ) );
+    assert( tr_isPeerIo( peerIo ) );
+
+    tr_ptrArrayInsertSorted( b->peers, peerIo, comparePointers );
+}
+
+void
+tr_bandwidthRemovePeer( tr_bandwidth  * b,
+                        tr_peerIo     * peerIo )
+{
+    assert( tr_isBandwidth( b ) );
+    assert( tr_isPeerIo( peerIo ) );
+
+    tr_ptrArrayRemoveSorted( b->peers, peerIo, comparePointers );
 }
 
@@ -361 +378 @@
                    size_t                byteCount )
 {
-    assert( isBandwidth( b ) );
-    assert( isDirection( dir ) );
+    assert( tr_isBandwidth( b ) );
+    assert( tr_isDirection( dir ) );
 
     if( b )
@@ -379 +396 @@
 tr_bandwidthGetRawSpeed( const tr_bandwidth * b, tr_direction dir )
 {
-    assert( isBandwidth( b ) );
-    assert( isDirection( dir ) );
+    assert( tr_isBandwidth( b ) );
+    assert( tr_isDirection( dir ) );
 
     return getSpeed( &b->band[dir].raw, HISTORY_MSEC );
@@ -388 +405 @@
 tr_bandwidthGetPieceSpeed( const tr_bandwidth * b, tr_direction dir )
 {
-    assert( isBandwidth( b ) );
-    assert( isDirection( dir ) );
+    assert( tr_isBandwidth( b ) );
+    assert( tr_isDirection( dir ) );
 
     return getSpeed( &b->band[dir].piece, HISTORY_MSEC );
@@ -398 +415 @@
                   tr_direction    dir,
                   size_t          byteCount,
-                  int             isPieceData )
+                  tr_bool         isPieceData )
 {
     struct tr_band * band;
     size_t oldBytesLeft;
 
-    assert( isBandwidth( b ) );
-    assert( isDirection( dir ) );
+    assert( tr_isBandwidth( b ) );
+    assert( tr_isDirection( dir ) );
 
     band = &b->band[dir];

branches/1.4x/libtransmission/bandwidth.h

@@ -1 +1 @@
 /*
- * This file Copyright (C) 2008 Charles Kerr <charles@rebelbase.com>
+ * This file Copyright (C) 2008 Charles Kerr <charles@transmissionbt.com>
  *
  * This file is licensed by the GPL version 2.  Works owned by the
@@ -59 +59 @@
  *   bandwidth they can safely use.
  */
+
 typedef struct tr_bandwidth tr_bandwidth;
+
+struct tr_peerIo;
 
 /**
@@ -72 +75 @@
 /** @brief destroy a tr_bandwidth object */
 void     tr_bandwidthFree             ( tr_bandwidth        * bandwidth );
+
+/** @brief test to see if the pointer refers to a live bandwidth object */
+tr_bool  tr_isBandwidth               ( const tr_bandwidth  * bandwidth );
 
 /******
@@ -98 +104 @@
 void    tr_bandwidthSetLimited        ( tr_bandwidth        * bandwidth,
                                         tr_direction          direction,
-                                        int                   isLimited );
+                                        tr_bool               isLimited );
 
 /**
  * @return nonzero if this bandwidth throttles its iobufs' speeds
  */
-int     tr_bandwidthIsLimited         ( const tr_bandwidth  * bandwidth,
+tr_bool tr_bandwidthIsLimited         ( const tr_bandwidth  * bandwidth,
                                         tr_direction          direction );
 
@@ -143 +149 @@
                                         tr_direction          direction,
                                         size_t                byteCount,
-                                        int                   isPieceData );
+                                        tr_bool               isPieceData );
 
 /******
@@ -160 +166 @@
 void    tr_bandwidthHonorParentLimits ( tr_bandwidth        * bandwidth,
                                         tr_direction          direction,
-                                        int                   isEnabled );
+                                        tr_bool               isEnabled );
 
 /******
@@ -167 +173 @@
 
 /**
- * @brief add an iobuf to this bandwidth's list of iobufs.
+ * @brief add a tr_peerIo to this bandwidth's list.
  * They will be notified when more bandwidth is made available for them to consume.
  */
-void    tr_bandwidthAddBuffer         ( tr_bandwidth        * bandwidth,
-                                        struct tr_iobuf     * iobuf );
+void    tr_bandwidthAddPeer           ( tr_bandwidth        * bandwidth,
+                                        struct tr_peerIo    * peerIo );
 
 /**
  * @brief remove an iobuf from this bandwidth's list of iobufs.
  */
-void    tr_bandwidthRemoveBuffer      ( tr_bandwidth        * bandwidth,
-                                        struct tr_iobuf     * iobuf );
+void    tr_bandwidthRemovePeer        ( tr_bandwidth        * bandwidth,
+                                        struct tr_peerIo    * peerIo );
 
 #endif

branches/1.4x/libtransmission/bencode-test.c

@@ -397 +397 @@
     int i;
 
-    if( ( i = testInt( ) ) )
+    if(( i = testInt( )))
         return i;
 
-    if( ( i = testStr( ) ) )
+    if(( i = testStr( )))
         return i;
 
-    if( ( i = testParse( ) ) )
+    if(( i = testParse( )))
         return i;
 
-    if( ( i = testJSON( ) ) )
+    if(( i = testJSON( )))
         return i;
 

branches/1.4x/libtransmission/fastresume.c

@@ -55 +55 @@
 #include "completion.h"
 #include "fastresume.h"
+#include "net.h"
 #include "peer-mgr.h"
 #include "platform.h"

branches/1.4x/libtransmission/handshake.c

@@ -1 +1 @@
 /*
- * This file Copyright (C) 2007-2008 Charles Kerr <charles@rebelbase.com>
+ * This file Copyright (C) 2007-2008 Charles Kerr <charles@transmissionbt.com>
  *
  * This file is licensed by the GPL version 2.  Works owned by the
@@ -25 +25 @@
 #include "crypto.h"
 #include "handshake.h"
-#include "iobuf.h"
 #include "peer-io.h"
 #include "peer-mgr.h"
@@ -34 +33 @@
 /* enable LibTransmission extension protocol */
 #define ENABLE_LTEP * /
-
-/* enable fast peers extension protocol */
-/* #define ENABLE_FASTPEER */
+/* fast extensions */
+#define ENABLE_FAST * /
 
 /***
@@ -73 +71 @@
 #endif
 
-#ifdef ENABLE_FASTPEER
- #define HANDSHAKE_HAS_FASTEXT( bits ) ( ( ( bits )[7] & 0x04 ) ? 1 : 0 )
- #define HANDSHAKE_SET_FASTEXT( bits ) ( ( bits )[7] |= 0x04 )
-#else
- #define HANDSHAKE_HAS_FASTEXT( bits ) ( 0 )
- #define HANDSHAKE_SET_FASTEXT( bits ) ( (void)0 )
-#endif
+#ifdef ENABLE_FAST 
+ #define HANDSHAKE_HAS_FASTEXT( bits ) ( ( ( bits )[7] & 0x04 ) ? 1 : 0 ) 
+ #define HANDSHAKE_SET_FASTEXT( bits ) ( ( bits )[7] |= 0x04 ) 
+#else 
+ #define HANDSHAKE_HAS_FASTEXT( bits ) ( 0 ) 
+ #define HANDSHAKE_SET_FASTEXT( bits ) ( (void)0 ) 
+#endif 
 
 /* http://www.azureuswiki.com/index.php/Extension_negotiation_protocol
@@ -94 +92 @@
     tr_peerIo *           io;
     tr_crypto *           crypto;
-    struct tr_handle *    handle;
+    tr_session *          session;
     uint8_t               myPublicKey[KEY_LEN];
     uint8_t               mySecret[KEY_LEN];
@@ -208 +206 @@
     uint8_t          * walk = buf;
     const uint8_t    * torrentHash = tr_cryptoGetTorrentHash( handshake->crypto );
-    const tr_torrent * tor = tr_torrentFindFromHash( handshake->handle, torrentHash );
+    const tr_torrent * tor = tr_torrentFindFromHash( handshake->session, torrentHash );
     const uint8_t    * peer_id = tor && tor->peer_id ? tor->peer_id : tr_getPeerId( );
 
@@ -250 +248 @@
     const uint8_t * peer_id;
 
-    dbgmsg( handshake, "payload: need %d, got %d", (int)HANDSHAKE_SIZE,
-           (int)EVBUFFER_LENGTH( inbuf ) );
+    dbgmsg( handshake, "payload: need %d, got %zu",
+            (int)HANDSHAKE_SIZE, EVBUFFER_LENGTH( inbuf ) );
 
     if( EVBUFFER_LENGTH( inbuf ) < HANDSHAKE_SIZE )
@@ -267 +265 @@
     tr_peerIoReadBytes( handshake->io, inbuf, hash, sizeof( hash ) );
     assert( tr_peerIoHasTorrentHash( handshake->io ) );
-    if( !tr_torrentExists( handshake->handle, hash )
+    if( !tr_torrentExists( handshake->session, hash )
       || memcmp( hash, tr_peerIoGetTorrentHash( handshake->io ),
                  SHA_DIGEST_LENGTH ) )
@@ -285 +283 @@
             handshake->peer_id );
 
-    tor = tr_torrentFindFromHash( handshake->handle, hash );
+    tor = tr_torrentFindFromHash( handshake->session, hash );
     peer_id = tor && tor->peer_id ? tor->peer_id : tr_getPeerId( );
     if( !memcmp( handshake->peer_id, peer_id, PEER_ID_LEN ) )
@@ -297 +295 @@
     **/
 
-    if( HANDSHAKE_HAS_LTEP( reserved ) )
-    {
-        tr_peerIoEnableLTEP( handshake->io, 1 );
-        dbgmsg( handshake, "using ltep" );
-    }
-
-    if( HANDSHAKE_HAS_FASTEXT( reserved ) )
-    {
-        tr_peerIoEnableFEXT( handshake->io, 1 );
-        dbgmsg( handshake, "using fext" );
-    }
+    tr_peerIoEnableLTEP( handshake->io, HANDSHAKE_HAS_LTEP( reserved ) );
+
+    tr_peerIoEnableFEXT( handshake->io, HANDSHAKE_HAS_FASTEXT( reserved ) );
 
     return HANDSHAKE_OK;
@@ -410 +400 @@
         return READ_LATER;
 
-    isEncrypted = memcmp( EVBUFFER_DATA(
-                              inbuf ), HANDSHAKE_NAME, HANDSHAKE_NAME_LEN );
+    isEncrypted = memcmp( EVBUFFER_DATA( inbuf ), HANDSHAKE_NAME, HANDSHAKE_NAME_LEN );
     if( isEncrypted )
     {
@@ -422 +411 @@
            ( isEncrypted ? "encrypted" : "plaintext" ) );
 
-    tr_peerIoSetEncryption(
-        handshake->io, isEncrypted ? PEER_ENCRYPTION_RC4
-        : PEER_ENCRYPTION_NONE );
+    tr_peerIoSetEncryption( handshake->io, isEncrypted ? PEER_ENCRYPTION_RC4
+                                                       : PEER_ENCRYPTION_NONE );
     if( !isEncrypted )
     {
@@ -454 +442 @@
         uint8_t buf[SHA_DIGEST_LENGTH];
         tr_sha1( req2, "req2", 4,
-                 tr_cryptoGetTorrentHash(
-                     handshake->crypto ), SHA_DIGEST_LENGTH, NULL );
+                 tr_cryptoGetTorrentHash( handshake->crypto ),
+                 SHA_DIGEST_LENGTH, NULL );
         tr_sha1( req3, "req3", 4, secret, KEY_LEN, NULL );
         for( i = 0; i < SHA_DIGEST_LENGTH; ++i )
@@ -576 +564 @@
     uint8_t *    tmp;
 
-    dbgmsg( handshake, "pad d: need %d, got %d", (int)needlen,
-           (int)EVBUFFER_LENGTH(
-               inbuf ) );
+    dbgmsg( handshake, "pad d: need %zu, got %zu",
+            needlen, EVBUFFER_LENGTH( inbuf ) );
     if( EVBUFFER_LENGTH( inbuf ) < needlen )
         return READ_LATER;
@@ -608 +595 @@
     uint8_t   hash[SHA_DIGEST_LENGTH];
 
-    dbgmsg( handshake, "payload: need %d, got %d",
-           (int)INCOMING_HANDSHAKE_LEN, (int)EVBUFFER_LENGTH(
-               inbuf ) );
+    dbgmsg( handshake, "payload: need %d, got %zu",
+            (int)INCOMING_HANDSHAKE_LEN, EVBUFFER_LENGTH( inbuf ) );
 
     if( EVBUFFER_LENGTH( inbuf ) < INCOMING_HANDSHAKE_LEN )
@@ -667 +653 @@
 
     /**
-    *** Extension negotiation
+    *** Extensions
     **/
 
-    if( HANDSHAKE_HAS_LTEP( reserved ) )
-    {
-        tr_peerIoEnableLTEP( handshake->io, 1 );
-        dbgmsg( handshake, "using ltep" );
-    }
-    if( HANDSHAKE_HAS_FASTEXT( reserved ) )
-    {
-        tr_peerIoEnableFEXT( handshake->io, 1 );
-        dbgmsg( handshake, "using fext" );
-    }
+    tr_peerIoEnableLTEP( handshake->io, HANDSHAKE_HAS_LTEP( reserved ) );
+
+    tr_peerIoEnableFEXT( handshake->io, HANDSHAKE_HAS_FASTEXT( reserved ) );
 
     /* torrent hash */
@@ -685 +664 @@
     if( tr_peerIoIsIncoming( handshake->io ) )
     {
-        if( !tr_torrentExists( handshake->handle, hash ) )
+        if( !tr_torrentExists( handshake->session, hash ) )
         {
             dbgmsg( handshake, "peer is trying to connect to us for a torrent we don't have." );
@@ -737 +716 @@
 
     /* peer id */
-    tr_peerIoReadBytes( handshake->io, inbuf, handshake->peer_id,
-                        PEER_ID_LEN );
+    tr_peerIoReadBytes( handshake->io, inbuf, handshake->peer_id, PEER_ID_LEN );
     tr_peerIoSetPeersId( handshake->io, handshake->peer_id );
     handshake->havePeerID = TRUE;
     tr_clientForId( client, sizeof( client ), handshake->peer_id );
     dbgmsg( handshake, "peer-id is [%s] ... isIncoming is %d", client,
-           tr_peerIoIsIncoming(
-               handshake->io ) );
+            tr_peerIoIsIncoming( handshake->io ) );
 
     /* if we've somehow connected to ourselves, don't keep the connection */
-    tor = tr_torrentFindFromHash( handshake->handle, tr_peerIoGetTorrentHash( handshake->io ) );
+    tor = tr_torrentFindFromHash( handshake->session, tr_peerIoGetTorrentHash( handshake->io ) );
     peer_id = tor && tor->peer_id ? tor->peer_id : tr_getPeerId( );
     peerIsGood = memcmp( handshake->peer_id, peer_id, PEER_ID_LEN ) != 0;
@@ -763 +740 @@
     int            len;
 
-    dbgmsg( handshake, "in readYa... need %d, have %d", (int)KEY_LEN,
-           (int)EVBUFFER_LENGTH(
-               inbuf ) );
+    dbgmsg( handshake, "in readYa... need %d, have %zu",
+            (int)KEY_LEN, EVBUFFER_LENGTH( inbuf ) );
     if( EVBUFFER_LENGTH( inbuf ) < KEY_LEN )
         return READ_LATER;
@@ -796 +772 @@
     uint8_t * pch;
 
-    dbgmsg(
-         handshake,
-        "looking to get past pad a... & resync on hash('req',S) ... have %d bytes",
-        (int)EVBUFFER_LENGTH( inbuf ) );
+    dbgmsg( handshake, "looking to get past pad a... & resync on hash('req',S) ... have %zu bytes",
+            EVBUFFER_LENGTH( inbuf ) );
     /**
     *** Resynchronizing on HASH('req1',S)
@@ -809 +783 @@
     if( pch == NULL )
     {
-        dbgmsg( handshake, "no luck so far.. draining %d bytes",
-               (int)EVBUFFER_LENGTH(
-                   inbuf ) );
+        dbgmsg( handshake, "no luck so far.. draining %zu bytes",
+                EVBUFFER_LENGTH( inbuf ) );
         evbuffer_drain( inbuf, EVBUFFER_LENGTH( inbuf ) );
         return READ_LATER;
@@ -871 +844 @@
         obfuscatedTorrentHash[i] = req2[i] ^ req3[i];
     if( ( tor =
-             tr_torrentFindFromObfuscatedHash( handshake->handle,
+             tr_torrentFindFromObfuscatedHash( handshake->session,
                                                obfuscatedTorrentHash ) ) )
     {
-        dbgmsg(
-            handshake,
-            "got INCOMING connection's encrypted handshake for torrent [%s]",
-            tor->info.name );
+        dbgmsg( handshake, "got INCOMING connection's encrypted handshake for torrent [%s]",
+                tor->info.name );
         tr_peerIoSetTorrentHash( handshake->io, tor->info.hash );
         if( !tr_torrentAllowsPex( tor )
-          && tr_peerMgrPeerIsSeed( handshake->handle->peerMgr,
+          && tr_peerMgrPeerIsSeed( handshake->session->peerMgr,
                                   tor->info.hash,
                                   tr_peerIoGetAddress( handshake->io, NULL ) ) )
@@ -938 +909 @@
     uint32_t          crypto_select;
 
-    dbgmsg( handshake, "reading IA... have %d, need %d",
-            (int)EVBUFFER_LENGTH(
-                inbuf ), (int)needlen );
+    dbgmsg( handshake, "reading IA... have %zu, need %zu",
+            EVBUFFER_LENGTH( inbuf ), needlen );
     if( EVBUFFER_LENGTH( inbuf ) < needlen )
         return READ_LATER;
 
     /**
-    ***  B->A: ENCRYPT(VC, crypto_select, len(padD), padD), ENCRYPT2(Payload
-    ***Stream)
+    ***  B->A: ENCRYPT(VC, crypto_select, len(padD), padD), ENCRYPT2(Payload Stream)
     **/
 
@@ -975 +944 @@
 
     dbgmsg( handshake, "sending pad d" );
-    /* ENCRYPT(VC, crypto_provide, len(PadC), PadC
+    /* ENCRYPT(VC, crypto_provide, len(PadD), PadD
      * PadD is reserved for future extensions to the handshake...
      * standard practice at this time is for it to be zero-length */
@@ -1013 +982 @@
     const size_t      needlen = HANDSHAKE_SIZE;
 
-    dbgmsg( handshake, "reading payload stream... have %d, need %d",
-            (int)EVBUFFER_LENGTH( inbuf ), (int)needlen );
+    dbgmsg( handshake, "reading payload stream... have %zu, need %zu",
+            EVBUFFER_LENGTH( inbuf ), needlen );
     if( EVBUFFER_LENGTH( inbuf ) < needlen )
         return READ_LATER;
@@ -1035 +1004 @@
 
 static ReadState
-canRead( struct tr_iobuf * iobuf, void * arg, size_t * piece )
-{
-    tr_handshake *    handshake = arg;
-    struct evbuffer * inbuf = tr_iobuf_input( iobuf );
+canRead( struct tr_peerIo * io, void * arg, size_t * piece )
+{
+    tr_handshake    * handshake = arg;
+    struct evbuffer * inbuf = tr_peerIoGetReadBuffer( io );
     ReadState         ret;
-    int               readyForMore = TRUE;
+    tr_bool           readyForMore = TRUE;
 
     /* no piece data in handshake */
@@ -1122 +1091 @@
 }
 
+void
+tr_handshakeFree( tr_handshake * handshake )
+{
+    if( handshake->io )
+        tr_peerIoFree( handshake->io );
+
+    tr_free( handshake );
+}
+
 static int
 tr_handshakeDone( tr_handshake * handshake,
                   int            isOK )
 {
-    int success;
+    tr_bool success;
 
     dbgmsg( handshake, "handshakeDone: %s", isOK ? "connected" : "aborting" );
@@ -1133 +1111 @@
     success = fireDoneFunc( handshake, isOK );
 
-    tr_free( handshake );
-
     return success ? READ_LATER : READ_ERR;
 }
@@ -1145 +1121 @@
 
 static void
-gotError( struct tr_iobuf  * iobuf UNUSED,
-          short              what,
-          void             * arg )
+gotError( tr_peerIo  * io UNUSED,
+          short        what,
+          void       * arg )
 {
     tr_handshake * handshake = (tr_handshake *) arg;
@@ -1194 +1170 @@
     handshake->doneCB = doneCB;
     handshake->doneUserData = doneUserData;
-    handshake->handle = tr_peerIoGetSession( io );
-    tr_peerIoSetTimeoutSecs( io, 15 );
+    handshake->session = tr_peerIoGetSession( io );
 
     tr_peerIoSetIOFuncs( handshake->io, canRead, NULL, gotError, handshake );
@@ -1225 +1200 @@
 }
 
-const struct in_addr *
-tr_handshakeGetAddr( const struct tr_handshake * handshake,
-                                                 uint16_t * port )
-{
+struct tr_peerIo*
+tr_handshakeStealIO( tr_handshake * handshake )
+{
+    struct tr_peerIo * io;
+
     assert( handshake );
     assert( handshake->io );
 
+    io = handshake->io;
+    handshake->io = NULL;
+    return io;
+}
+
+const tr_address *
+tr_handshakeGetAddr( const struct tr_handshake * handshake,
+                     tr_port                   * port )
+{
+    assert( handshake );
+    assert( handshake->io );
+
     return tr_peerIoGetAddress( handshake->io, port );
 }

branches/1.4x/libtransmission/handshake.h

@@ -1 +1 @@
 /*
- * This file Copyright (C) 2007-2008 Charles Kerr <charles@rebelbase.com>
+ * This file Copyright (C) 2007-2008 Charles Kerr <charles@transmissionbt.com>
  *
  * This file is licensed by the GPL version 2.  Works owned by the
@@ -19 +19 @@
 
 #include "transmission.h"
+#include "net.h"
 
-struct in_addr;
 struct tr_peerIo;
 typedef struct tr_handshake tr_handshake;
 
 /* returns true on success, false on error */
-typedef int ( *handshakeDoneCB )( struct tr_handshake * handshake,
-                                  struct tr_peerIo *    io,
-                                  int                   isConnected,
-                                  const uint8_t *       peerId,
-                                  void *                userData );
+typedef tr_bool ( *handshakeDoneCB )( struct tr_handshake * handshake,
+                                      struct tr_peerIo *    io,
+                                      int                   isConnected,
+                                      const uint8_t *       peerId,
+                                      void *                userData );
 
 tr_handshake *         tr_handshakeNew( struct tr_peerIo * io,
@@ -36 +36 @@
                                         void *             doneUserData );
 
-const struct in_addr * tr_handshakeGetAddr(
-    const struct tr_handshake * handshake,
-                                uint16_t
-    *                           setme_port );
+const tr_address *     tr_handshakeGetAddr( const struct tr_handshake  * handshake,
+                                            tr_port                    * port );
+
+void                   tr_handshakeFree( tr_handshake * handshake );
 
 void                   tr_handshakeAbort( tr_handshake * handshake );
@@ -45 +45 @@
 struct tr_peerIo*      tr_handshakeGetIO( tr_handshake * handshake );
 
+struct tr_peerIo*      tr_handshakeStealIO( tr_handshake * handshake );
+
+
 #endif

branches/1.4x/libtransmission/net.c

@@ -64 +64 @@
 }
 
+tr_bool
+tr_isAddress( const tr_address * a )
+{
+    return a != NULL; /* this is implemented better in 1.50 */
+}
+
+
 /***********************************************************************
  * DNS resolution
@@ -243 +250 @@
 }
 
+int
+tr_compareAddresses( const struct in_addr * a, const struct in_addr * b )
+{
+    if( a->s_addr != b->s_addr )
+        return a->s_addr < b->s_addr ? -1 : 1;
+
+    return 0;
+}

branches/1.4x/libtransmission/net.h

@@ -65 +65 @@
 struct tr_session;
 
+typedef struct in_addr tr_address;
+
+tr_bool tr_isAddress( const tr_address * a );
+
 /***********************************************************************
  * DNS resolution
@@ -95 +99 @@
                  int                    len );
 
+int tr_compareAddresses( const struct in_addr * a,
+                         const struct in_addr * b );
+
 void tr_netInit( void );
 

branches/1.4x/libtransmission/peer-common.h

@@ -23 +23 @@
 
 #include "transmission.h"
-#include "publish.h"
 
 typedef enum
@@ -43 +42 @@
     TR_PEER_CLIENT_GOT_BLOCK,
     TR_PEER_CLIENT_GOT_DATA,
+    TR_PEER_CLIENT_GOT_ALLOWED_FAST,
+    TR_PEER_CLIENT_GOT_SUGGEST,
     TR_PEER_PEER_GOT_DATA,
     TR_PEER_PEER_PROGRESS,
     TR_PEER_ERROR,
     TR_PEER_CANCEL,
+    TR_PEER_UPLOAD_ONLY,
     TR_PEER_NEED_REQ
 }
@@ -54 +56 @@
 {
     PeerEventType    eventType;
-    uint32_t         pieceIndex;   /* for GOT_BLOCK, CANCEL */
+    uint32_t         pieceIndex;   /* for GOT_BLOCK, CANCEL, ALLOWED, SUGGEST */
     uint32_t         offset;       /* for GOT_BLOCK */
     uint32_t         length;       /* for GOT_BLOCK + GOT_DATA */
     float            progress;     /* for PEER_PROGRESS */
     int              err;          /* errno for GOT_ERROR */
-    int              wasPieceData; /* for GOT_DATA */
+    tr_bool          wasPieceData; /* for GOT_DATA */
+    tr_bool          uploadOnly;   /* for UPLOAD_ONLY */
 }
 tr_peer_event;

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

@@ -1 +1 @@
 /*
- * This file Copyright (C) 2007-2008 Charles Kerr <charles@rebelbase.com>
+ * This file Copyright (C) 2007-2008 Charles Kerr <charles@transmissionbt.com>
  *
  * This file is licensed by the GPL version 2.  Works owned by the
@@ -20 +20 @@
  #include <winsock2.h>
 #else
- #include <netinet/in.h> /* struct in_addr */
  #include <arpa/inet.h> /* inet_ntoa */
 #endif
@@ -29 +28 @@
 #include "bandwidth.h"
 #include "crypto.h"
-#include "iobuf.h"
 #include "list.h"
 #include "net.h"
@@ -37 +35 @@
 
 #define MAGIC_NUMBER 206745
-#define IO_TIMEOUT_SECS 8
 
 static size_t
@@ -81 +78 @@
 struct tr_peerIo
 {
-    tr_bool                  isEncrypted;
-    tr_bool                  isIncoming;
-    tr_bool                  peerIdIsSet;
-    tr_bool                  extendedProtocolSupported;
-    tr_bool                  fastPeersSupported;
-
-    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;
+    tr_bool            isEncrypted;
+    tr_bool            isIncoming;
+    tr_bool            peerIdIsSet;
+    tr_bool            extendedProtocolSupported;
+    tr_bool            fastExtensionSupported;
+
+    int                magicNumber;
+
+    uint8_t            encryptionMode;
+    tr_port            port;
+    int                socket;
+
+    uint8_t            peerId[20];
+    time_t             timeCreated;
+
+    tr_session       * session;
+
+    tr_address         addr;
+    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;
+
+    struct evbuffer  * inbuf;
+    struct evbuffer  * outbuf;
+
+    struct event       event_read;
+    struct event       event_write;
 };
 
@@ -119 +120 @@
 
 static void
-didWriteWrapper( struct tr_iobuf  * iobuf,
-                 size_t             bytes_transferred,
-                 void             * vio )
-{
-    tr_peerIo *  io = vio;
-
+didWriteWrapper( tr_peerIo * io, size_t bytes_transferred )
+{
     while( bytes_transferred )
     {
@@ -144 +141 @@
             tr_free( tr_list_pop_front( &io->output_datatypes ) );
     }
-
-    if( EVBUFFER_LENGTH( tr_iobuf_output( iobuf ) ) )
-        tr_iobuf_enable( io->iobuf, EV_WRITE );
 }
 
 static void
-canReadWrapper( struct tr_iobuf  * iobuf,
-                size_t             bytes_transferred UNUSED,
-                void              * vio )
-{
-    int          done = 0;
-    int          err = 0;
-    tr_peerIo *  io = vio;
+canReadWrapper( tr_peerIo * io )
+{
+    tr_bool done = 0;
+    tr_bool err = 0;
     tr_session * session = io->session;
 
@@ -169 +160 @@
         {
             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 != READ_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 );
-            }
+            const size_t oldLen = EVBUFFER_LENGTH( io->inbuf );
+            const int ret = io->canRead( io, io->userData, &piece );
+
+            const size_t used = oldLen - EVBUFFER_LENGTH( io->inbuf );
+
+            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( tr_iobuf_input( iobuf )))
+                    if( EVBUFFER_LENGTH( io->inbuf ) )
                         continue;
                     done = 1;
@@ -203 +193 @@
 }
 
+#define _isBool(b) (((b)==0 || (b)==1))
+
+tr_bool
+tr_isPeerIo( const tr_peerIo * io )
+{
+    return ( io != NULL )
+        && ( io->magicNumber == MAGIC_NUMBER )
+        && ( _isBool( io->isEncrypted ) )
+        && ( _isBool( io->isIncoming ) )
+        && ( _isBool( io->peerIdIsSet ) )
+        && ( _isBool( io->extendedProtocolSupported ) )
+        && ( _isBool( io->fastExtensionSupported ) );
+}
+
 static void
-gotErrorWrapper( struct tr_iobuf  * iobuf,
-                 short              what,
-                 void             * userData )
-{
-    tr_peerIo * c = userData;
-
-    if( c->gotError )
-        c->gotError( iobuf, what, c->userData );
+event_read_cb( int fd, short event UNUSED, void * vio )
+{
+    int res;
+    short what = EVBUFFER_READ;
+    tr_peerIo * io = vio;
+    const size_t howmuch = tr_bandwidthClamp( io->bandwidth, TR_DOWN, io->session->so_rcvbuf );
+    const tr_direction dir = TR_DOWN;
+
+    assert( tr_isPeerIo( io ) );
+
+    dbgmsg( io, "libevent says this peer is ready to read" );
+
+    /* if we don't have any bandwidth left, stop reading */
+    if( howmuch < 1 ) {
+        tr_peerIoSetEnabled( io, dir, FALSE );
+        return;
+    }
+
+    res = evbuffer_read( io->inbuf, fd, howmuch );
+    if( res == -1 ) {
+        if( errno == EAGAIN || errno == EINTR )
+            goto reschedule;
+        /* error case */
+        what |= EVBUFFER_ERROR;
+    } else if( res == 0 ) {
+        /* eof case */
+        what |= EVBUFFER_EOF;
+    }
+
+    if( res <= 0 )
+        goto error;
+
+    tr_peerIoSetEnabled( io, dir, TRUE );
+
+    /* Invoke the user callback - must always be called last */
+    canReadWrapper( io );
+
+    return;
+
+ reschedule:
+    tr_peerIoSetEnabled( io, dir, TRUE );
+    return;
+
+ error:
+    if( io->gotError != NULL )
+        io->gotError( io, what, io->userData );
+}
+
+static int
+tr_evbuffer_write( tr_peerIo * io, int fd, size_t howmuch )
+{
+    struct evbuffer * buffer = io->outbuf;
+    int n = MIN( EVBUFFER_LENGTH( buffer ), howmuch );
+
+#ifdef WIN32
+    n = send(fd, buffer->buffer, n,  0 );
+#else
+    n = write(fd, buffer->buffer, n );
+#endif
+    dbgmsg( io, "wrote %d to peer (%s)", n, (n==-1?strerror(errno):"") );
+
+    if( n == -1 )
+        return -1;
+    if (n == 0)
+        return 0;
+    evbuffer_drain( buffer, n );
+
+    return n;
+}
+
+static void
+event_write_cb( int fd, short event UNUSED, void * vio )
+{
+    int res = 0;
+    short what = EVBUFFER_WRITE;
+    tr_peerIo * io = vio;
+    size_t howmuch;
+    const tr_direction dir = TR_UP;
+
+    assert( tr_isPeerIo( io ) );
+
+    dbgmsg( io, "libevent says this peer is ready to write" );
+
+    howmuch = MIN( (size_t)io->session->so_sndbuf, EVBUFFER_LENGTH( io->outbuf ) );
+    howmuch = tr_bandwidthClamp( io->bandwidth, dir, howmuch );
+
+    /* if we don't have any bandwidth left, stop writing */
+    if( howmuch < 1 ) {
+        tr_peerIoSetEnabled( io, dir, FALSE );
+        return;
+    }
+
+    res = tr_evbuffer_write( io, fd, howmuch );
+    if (res == -1) {
+#ifndef WIN32
+/*todo. evbuffer uses WriteFile when WIN32 is set. WIN32 system calls do not
+ *  *set errno. thus this error checking is not portable*/
+        if (errno == EAGAIN || errno == EINTR || errno == EINPROGRESS)
+            goto reschedule;
+        /* error case */
+        what |= EVBUFFER_ERROR;
+
+#else
+        goto reschedule;
+#endif
+
+    } else if (res == 0) {
+        /* eof case */
+        what |= EVBUFFER_EOF;
+    }
+    if (res <= 0)
+        goto error;
+
+    if( EVBUFFER_LENGTH( io->outbuf ) )
+        tr_peerIoSetEnabled( io, dir, TRUE );
+
+    didWriteWrapper( io, res );
+    return;
+
+ reschedule:
+    if( EVBUFFER_LENGTH( io->outbuf ) )
+        tr_peerIoSetEnabled( io, dir, TRUE );
+    return;
+
+ error:
+    if( io->gotError != NULL )
+        io->gotError( io, what, io->userData );
 }
 
@@ -218 +341 @@
 **/
 
-static void
-bufevNew( tr_peerIo * io )
-{
-    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 );
+isFlag( int flag )
+{
+    return( ( flag == 0 ) || ( flag == 1 ) );
 }
 
 static tr_peerIo*
-tr_peerIoNew( tr_session *           session,
-              const struct in_addr * in_addr,
-              uint16_t               port,
-              const uint8_t *        torrentHash,
-              int                    isIncoming,
-              int                    socket )
+tr_peerIoNew( tr_session       * session,
+              const tr_address * addr,
+              tr_port            port,
+              const uint8_t    * torrentHash,
+              int                isIncoming,
+              int                socket )
 {
     tr_peerIo * io;
@@ -256 +365 @@
     io->crypto = tr_cryptoNew( torrentHash, isIncoming );
     io->session = session;
-    io->in_addr = *in_addr;
+    io->addr = *addr;
     io->port = port;
     io->socket = socket;
     io->isIncoming = isIncoming != 0;
-    io->timeout = IO_TIMEOUT_SECS;
     io->timeCreated = time( NULL );
+    io->inbuf = evbuffer_new( );
+    io->outbuf = evbuffer_new( );
+    event_set( &io->event_read, io->socket, EV_READ, event_read_cb, io );
+    event_set( &io->event_write, io->socket, EV_WRITE, event_write_cb, io );
+#if 0
     bufevNew( io );
+#endif
     tr_peerIoSetBandwidth( io, session->bandwidth );
     return io;
@@ -268 +382 @@
 
 tr_peerIo*
-tr_peerIoNewIncoming( tr_session *           session,
-                      const struct in_addr * in_addr,
-                      uint16_t               port,
-                      int                    socket )
+tr_peerIoNewIncoming( tr_session       * session,
+                      const tr_address * addr,
+                      tr_port            port,
+                      int                socket )
 {
     assert( session );
-    assert( in_addr );
+    assert( addr );
     assert( socket >= 0 );
 
-    return tr_peerIoNew( session, in_addr, port,
-                         NULL, 1,
-                         socket );
+    return tr_peerIoNew( session, addr, port, NULL, 1, socket );
 }
 
 tr_peerIo*
-tr_peerIoNewOutgoing( tr_session *           session,
-                      const struct in_addr * in_addr,
-                      int                    port,
-                      const uint8_t *        torrentHash )
+tr_peerIoNewOutgoing( tr_session       * session,
+                      const tr_address * addr,
+                      tr_port            port,
+                      const uint8_t    * torrentHash )
 {
     int socket;
 
     assert( session );
-    assert( in_addr );
-    assert( port >= 0 );
+    assert( addr );
     assert( torrentHash );
 
-    socket = tr_netOpenTCP( session, in_addr, port );
+    socket = tr_netOpenTCP( session, addr, port );
 
     return socket < 0
            ? NULL
-           : tr_peerIoNew( session, in_addr, port, torrentHash, 0, socket );
+           : tr_peerIoNew( session, addr, port, torrentHash, 0, socket );
 }
 
@@ -307 +418 @@
     tr_peerIo * io = vio;
 
+    event_del( &io->event_read );
+    event_del( &io->event_write );
     tr_peerIoSetBandwidth( io, NULL );
-    tr_iobuf_free( io->iobuf );
+    evbuffer_free( io->outbuf );
+    evbuffer_free( io->inbuf );
     tr_netClose( io->socket );
     tr_cryptoFree( io->crypto );
@@ -332 +446 @@
 tr_peerIoGetSession( tr_peerIo * io )
 {
-    assert( isPeerIo( io ) );
+    assert( tr_isPeerIo( io ) );
     assert( io->session );
 
@@ -338 +452 @@
 }
 
-const struct in_addr*
+const tr_address*
 tr_peerIoGetAddress( const tr_peerIo * io,
-                           uint16_t * port )
-{
-    assert( isPeerIo( io ) );
+                           tr_port   * port )
+{
+    assert( tr_isPeerIo( io ) );
 
     if( port )
         *port = io->port;
 
-    return &io->in_addr;
+    return &io->addr;
 }
 
 const char*
-tr_peerIoAddrStr( const struct in_addr * addr,
-                  uint16_t               port )
+tr_peerIoAddrStr( const tr_address * addr, tr_port port )
 {
     static char buf[512];
-
-    tr_snprintf( buf, sizeof( buf ), "%s:%u", inet_ntoa( *addr ),
-                ntohs( port ) );
+    tr_snprintf( buf, sizeof( buf ), "%s:%u", inet_ntoa( *addr ), ntohs( port ) );
     return buf;
 }
@@ -364 +475 @@
 tr_peerIoGetAddrStr( const tr_peerIo * io )
 {
-    return tr_peerIoAddrStr( &io->in_addr, io->port );
-}
-
-static void
-tr_peerIoTryRead( tr_peerIo * io )
-{
-    if( EVBUFFER_LENGTH( tr_iobuf_input( io->iobuf )))
-        (*canReadWrapper)( io->iobuf, ~0, io );
-}
-
-void
-tr_peerIoSetIOFuncs( tr_peerIo *     io,
-                     tr_can_read_cb  readcb,
-                     tr_did_write_cb writecb,
-                     tr_net_error_cb errcb,
-                     void *          userData )
+    return tr_peerIoAddrStr( &io->addr, io->port );
+}
+
+void
+tr_peerIoSetIOFuncs( tr_peerIo        * io,
+                     tr_can_read_cb     readcb,
+                     tr_did_write_cb    writecb,
+                     tr_net_error_cb    errcb,
+                     void             * userData )
 {
     io->canRead = readcb;
@@ -385 +489 @@
     io->gotError = errcb;
     io->userData = userData;
-
-    tr_peerIoTryRead( io );
-}
-
-int
+}
+
+tr_bool
 tr_peerIoIsIncoming( const tr_peerIo * c )
 {
-    return c->isIncoming ? 1 : 0;
+    return c->isIncoming != 0;
 }
 
@@ -403 +505 @@
         tr_netClose( io->socket );
 
-    io->socket = tr_netOpenTCP( io->session, &io->in_addr, io->port );
+    io->socket = tr_netOpenTCP( io->session, &io->addr, io->port );
 
     if( io->socket >= 0 )
@@ -411 +513 @@
 
         tr_netSetTOS( io->socket, io->session->peerSocketTOS );
-        tr_iobuf_free( io->iobuf );
+#if 0
         bufevNew( io );
+#endif
 
         tr_peerIoSetBandwidth( io, bandwidth );
@@ -419 +522 @@
 
     return -1;
-}
-
-void
-tr_peerIoSetTimeoutSecs( tr_peerIo * io,
-                         int         secs )
-{
-    io->timeout = secs;
-    tr_iobuf_settimeout( io->iobuf, io->timeout, io->timeout );
-    tr_iobuf_enable( io->iobuf, EV_READ | EV_WRITE );
 }
 
@@ -438 +532 @@
                          const uint8_t * hash )
 {
-    assert( isPeerIo( io ) );
+    assert( tr_isPeerIo( io ) );
 
     tr_cryptoSetTorrentHash( io->crypto, hash );
@@ -446 +540 @@
 tr_peerIoGetTorrentHash( tr_peerIo * io )
 {
-    assert( isPeerIo( io ) );
+    assert( tr_isPeerIo( io ) );
     assert( io->crypto );
 
@@ -455 +549 @@
 tr_peerIoHasTorrentHash( const tr_peerIo * io )
 {
-    assert( isPeerIo( io ) );
+    assert( tr_isPeerIo( io ) );
     assert( io->crypto );
 
@@ -469 +563 @@
                      const uint8_t * peer_id )
 {
-    assert( isPeerIo( io ) );
+    assert( tr_isPeerIo( io ) );
 
     if( ( io->peerIdIsSet = peer_id != NULL ) )
@@ -480 +574 @@
 tr_peerIoGetPeersId( const tr_peerIo * io )
 {
-    assert( isPeerIo( io ) );
+    assert( tr_isPeerIo( io ) );
     assert( io->peerIdIsSet );
 
@@ -491 +585 @@
 
 void
-tr_peerIoEnableLTEP( tr_peerIo * io,
-                     int         flag )
-{
-    assert( isPeerIo( io ) );
-    assert( flag == 0 || flag == 1 );
-
+tr_peerIoEnableFEXT( tr_peerIo * io,
+                     tr_bool     flag )
+{
+    assert( tr_isPeerIo( io ) );
+    assert( isFlag( flag ) );
+
+    dbgmsg( io, "setting FEXT support flag to %d", (flag?1:0) );
+    io->fastExtensionSupported = flag;
+}
+
+tr_bool
+tr_peerIoSupportsFEXT( const tr_peerIo * io )
+{
+    assert( tr_isPeerIo( io ) );
+
+    return io->fastExtensionSupported;
+}
+
+/**
+***
+**/
+
+void
+tr_peerIoEnableLTEP( tr_peerIo  * io,
+                     tr_bool      flag )
+{
+    assert( tr_isPeerIo( io ) );
+    assert( isFlag( flag ) );
+
+    dbgmsg( io, "setting LTEP support flag to %d", (flag?1:0) );
     io->extendedProtocolSupported = flag;
 }
 
-void
-tr_peerIoEnableFEXT( tr_peerIo * io,
-                     int         flag )
-{
-    assert( isPeerIo( io ) );
-    assert( flag == 0 || flag == 1 );
-
-    io->fastPeersSupported = flag;
-}
-
-int
+tr_bool
 tr_peerIoSupportsLTEP( const tr_peerIo * io )
 {
-    assert( isPeerIo( io ) );
+    assert( tr_isPeerIo( io ) );
 
     return io->extendedProtocolSupported;
-}
-
-int
-tr_peerIoSupportsFEXT( const tr_peerIo * io )
-{
-    assert( isPeerIo( io ) );
-
-    return io->fastPeersSupported;
 }
 
@@ -540 +640 @@
     const double currentSpeed = tr_bandwidthGetPieceSpeed( io->bandwidth, TR_UP );
     const double period = 20; /* arbitrary */
-    return MAX( maxBlockSize*5.5, currentSpeed*1024*period );
+    const double numBlocks = 5.5; /* the 5 is arbitrary; the .5 is to leave room for messages */
+    return MAX( maxBlockSize*numBlocks, currentSpeed*1024*period );
 }
 
@@ -547 +648 @@
 {
     const size_t desiredLen = getDesiredOutputBufferSize( io );
-    const size_t currentLen = EVBUFFER_LENGTH( tr_iobuf_output( io->iobuf ) );
+    const size_t currentLen = EVBUFFER_LENGTH( io->outbuf );
     size_t freeSpace = 0;
 
@@ -560 +661 @@
                        tr_bandwidth  * bandwidth )
 {
-    assert( isPeerIo( io ) );
+    assert( tr_isPeerIo( io ) );
 
     if( io->bandwidth )
-        tr_bandwidthRemoveBuffer( io->bandwidth, io->iobuf );
+        tr_bandwidthRemovePeer( io->bandwidth, io );
 
     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 );
+        tr_bandwidthAddPeer( io->bandwidth, io );
 }
 
@@ -588 +686 @@
                         int         encryptionMode )
 {
-    assert( isPeerIo( io ) );
+    assert( tr_isPeerIo( io ) );
     assert( encryptionMode == PEER_ENCRYPTION_NONE
           || encryptionMode == PEER_ENCRYPTION_RC4 );
@@ -620 +718 @@
     tr_list_append( &io->output_datatypes, datatype );
 
-    evbuffer_add( tr_iobuf_output( io->iobuf ), writeme, writemeLen );
-    tr_iobuf_enable( io->iobuf, EV_WRITE );
+    evbuffer_add( io->outbuf, writeme, writemeLen );
 }
 
@@ -736 +833 @@
     uint16_t tmp;
 
+    assert( tr_isPeerIo( io ) );
+
     tr_peerIoReadBytes( io, inbuf, &tmp, sizeof( uint16_t ) );
     *setme = ntohs( tmp );
@@ -747 +846 @@
     uint32_t tmp;
 
+    assert( tr_isPeerIo( io ) );
+
     tr_peerIoReadBytes( io, inbuf, &tmp, sizeof( uint32_t ) );
     *setme = ntohl( tmp );
@@ -756 +857 @@
                 size_t            byteCount )
 {
-    uint8_t * tmp = tr_new( uint8_t, byteCount );
-
+    uint8_t * tmp;
+
+    assert( tr_isPeerIo( io ) );
+
+    tmp = tr_new( uint8_t, byteCount );
     tr_peerIoReadBytes( io, inbuf, tmp, byteCount );
     tr_free( tmp );
@@ -767 +871 @@
     return time( NULL ) - io->timeCreated;
 }
+
+/***
+****
+***/
+
+static int
+tr_peerIoTryRead( tr_peerIo * io, size_t howmuch )
+{
+    int res;
+
+    assert( tr_isPeerIo( io ) );
+
+    howmuch = tr_bandwidthClamp( io->bandwidth, TR_DOWN, howmuch );
+
+    res = howmuch ? evbuffer_read( io->inbuf, io->socket, howmuch ) : 0;
+
+    dbgmsg( io, "read %d from peer (%s)", res, (res==-1?strerror(errno):"") );
+
+    if( EVBUFFER_LENGTH( io->inbuf ) )
+        canReadWrapper( io );
+
+    if( ( res <= 0 ) && ( io->gotError ) && ( errno != EAGAIN ) && ( errno != EINTR ) && ( errno != EINPROGRESS ) )
+    {
+        short what = EVBUFFER_READ | EVBUFFER_ERROR;
+        if( res == 0 )
+            what |= EVBUFFER_EOF;
+        io->gotError( io, what, io->userData );
+    }
+
+    return res;
+}
+
+static int
+tr_peerIoTryWrite( tr_peerIo * io, size_t howmuch )
+{
+    int n;
+
+    assert( tr_isPeerIo( io ) );
+
+    howmuch = tr_bandwidthClamp( io->bandwidth, TR_UP, howmuch );
+
+    n = tr_evbuffer_write( io, io->socket, (int)howmuch );
+
+    if( n > 0 )
+        didWriteWrapper( io, n );
+
+    if( ( n < 0 ) && ( io->gotError ) && ( errno != EPIPE ) && ( errno != EAGAIN ) && ( errno != EINTR ) && ( errno != EINPROGRESS ) ) {
+        short what = EVBUFFER_WRITE | EVBUFFER_ERROR;
+        io->gotError( io, what, io->userData );
+    }
+
+    return n;
+}
+
+int
+tr_peerIoFlush( tr_peerIo  * io, tr_direction dir, size_t limit )
+{
+    int ret;
+
+    assert( tr_isPeerIo( io ) );
+    assert( tr_isDirection( dir ) );
+
+    if( dir==TR_DOWN )
+        ret = tr_peerIoTryRead( io, limit );
+    else
+        ret = tr_peerIoTryWrite( io, limit );
+
+    return ret;
+}
+
+struct evbuffer *
+tr_peerIoGetReadBuffer( tr_peerIo * io )
+{
+    assert( tr_isPeerIo( io ) );
+
+    return io->inbuf;
+}
+
+tr_bool
+tr_peerIoHasBandwidthLeft( const tr_peerIo * io, tr_direction dir )
+{
+    assert( tr_isPeerIo( io ) );
+    assert( tr_isDirection( dir ) );
+
+    return tr_bandwidthClamp( io->bandwidth, dir, 1024 ) > 0;
+}
+
+/***
+****
+****/
+
+static void
+event_enable( tr_peerIo * io, short event )
+{
+    assert( tr_isPeerIo( io ) );
+
+    if( event & EV_READ )
+        event_add( &io->event_read, NULL );
+
+    if( event & EV_WRITE )
+        event_add( &io->event_write, NULL );
+}
+
+static void
+event_disable( struct tr_peerIo * io, short event )
+{
+    assert( tr_isPeerIo( io ) );
+
+    if( event & EV_READ )
+        event_del( &io->event_read );
+
+    if( event & EV_WRITE )
+        event_del( &io->event_write );
+}
+
+
+void
+tr_peerIoSetEnabled( tr_peerIo    * io,
+                     tr_direction   dir,
+                     tr_bool        isEnabled )
+{
+    short event;
+
+    assert( tr_isPeerIo( io ) );
+    assert( tr_isDirection( dir ) );
+
+    event = dir == TR_UP ? EV_WRITE : EV_READ;
+
+    if( isEnabled )
+        event_enable( io, event );
+    else
+        event_disable( io, event );
+}

branches/1.4x/libtransmission/peer-io.h

@@ -1 +1 @@
 /*
- * This file Copyright (C) 2007-2008 Charles Kerr <charles@rebelbase.com>
+ * This file Copyright (C) 2007-2008 Charles Kerr <charles@transmissionbt.com>
  *
  * This file is licensed by the GPL version 2.  Works owned by the
@@ -18 +18 @@
 #define TR_PEER_IO_H
 
-/**
-***
-**/
-
-struct in_addr;
+#include "net.h" /* tr_address */
+/**
+***
+**/
+
 struct evbuffer;
 struct tr_bandwidth;
 struct tr_crypto;
-struct tr_iobuf;
 typedef struct tr_peerIo tr_peerIo;
 
@@ -33 +32 @@
 **/
 
-tr_peerIo*           tr_peerIoNewOutgoing( struct tr_handle     * session,
-                                           const struct in_addr * addr,
-                                           int                    port,
-                                           const  uint8_t       * torrentHash );
-
-tr_peerIo*           tr_peerIoNewIncoming( struct tr_handle     * session,
-                                           const struct in_addr * addr,
-                                           uint16_t               port,
-                                           int                    socket );
-
-void                 tr_peerIoFree( tr_peerIo * io );
-
-tr_session*          tr_peerIoGetSession( tr_peerIo * io );
-
-/**
-***
-**/
-
-void                 tr_peerIoEnableLTEP( tr_peerIo * io,
-                                          int         flag );
-
-void                 tr_peerIoEnableFEXT( tr_peerIo * io,
-                                          int         flag );
-
-int                  tr_peerIoSupportsLTEP( const tr_peerIo * io );
-
-int                  tr_peerIoSupportsFEXT( const tr_peerIo * io );
-
-/**
-***
-**/
-
-const char*          tr_peerIoAddrStr( const struct in_addr * addr,
-                                       uint16_t               port );
-
-const char*          tr_peerIoGetAddrStr( const tr_peerIo * io );
-
-const struct in_addr*tr_peerIoGetAddress( const tr_peerIo * io,
-                                                uint16_t * port );
+tr_peerIo*  tr_peerIoNewOutgoing( tr_session        * session,
+                                  const tr_address  * addr,
+                                  tr_port             port,
+                                  const  uint8_t    * torrentHash );
+
+tr_peerIo*  tr_peerIoNewIncoming( tr_session        * session,
+                                  const tr_address  * addr,
+                                  tr_port             port,
+                                  int                 socket );
+
+void        tr_peerIoFree       ( tr_peerIo         * io );
+
+tr_bool     tr_isPeerIo         ( const tr_peerIo   * io );
+
+
+/**
+***
+**/
+
+void        tr_peerIoEnableLTEP( tr_peerIo * io, tr_bool flag );
+
+tr_bool     tr_peerIoSupportsLTEP( const tr_peerIo * io );
+
+void        tr_peerIoEnableFEXT( tr_peerIo * io, tr_bool flag );
+
+tr_bool     tr_peerIoSupportsFEXT( const tr_peerIo * io );
+
+/**
+***
+**/
+
+tr_session* tr_peerIoGetSession ( tr_peerIo * io );
+
+const char* tr_peerIoAddrStr( const tr_address * addr,
+                              tr_port            port );
+
+const char* tr_peerIoGetAddrStr( const tr_peerIo * io );
+
+const tr_address * tr_peerIoGetAddress( const tr_peerIo * io,
+                                        tr_port         * port );
 
 const uint8_t*       tr_peerIoGetTorrentHash( tr_peerIo * io );
@@ -82 +82 @@
 int                  tr_peerIoReconnect( tr_peerIo * io );
 
-int                  tr_peerIoIsIncoming( const tr_peerIo * io );
-
-void                 tr_peerIoSetTimeoutSecs( tr_peerIo * io,
-                                              int         secs );
+tr_bool              tr_peerIoIsIncoming( const tr_peerIo * io );
 
 int                  tr_peerIoGetAge( const tr_peerIo * io );
@@ -111 +108 @@
 ReadState;
 
-typedef ReadState ( *tr_can_read_cb  )( struct tr_iobuf  * iobuf,
+typedef ReadState ( *tr_can_read_cb  )( tr_peerIo        * io,
                                         void             * user_data,
                                         size_t           * setme_piece_byte_count );
@@ -120 +117 @@
                                         void             * userData );
 
-typedef void      ( *tr_net_error_cb )( struct tr_iobuf  * ev,
+typedef void      ( *tr_net_error_cb )( tr_peerIo        * io,
                                         short              what,
                                         void             * userData );
@@ -157 +154 @@
 EncryptionMode;
 
-void              tr_peerIoSetEncryption( tr_peerIo * io,
-                                          int         encryptionMode );
-
-int               tr_peerIoIsEncrypted( const tr_peerIo * io );
-
-void              tr_peerIoWriteBytes( tr_peerIo *       io,
-                                       struct evbuffer * outbuf,
-                                       const void *      bytes,
-                                       size_t            byteCount );
-
-void              tr_peerIoWriteUint8( tr_peerIo *       io,
-                                       struct evbuffer * outbuf,
-                                       uint8_t           writeme );
-
-void              tr_peerIoWriteUint16( tr_peerIo *       io,
-                                        struct evbuffer * outbuf,
-                                        uint16_t          writeme );
-
-void              tr_peerIoWriteUint32( tr_peerIo *       io,
-                                        struct evbuffer * outbuf,
-                                        uint32_t          writeme );
-
-void              tr_peerIoReadBytes( tr_peerIo *       io,
-                                      struct evbuffer * inbuf,
-                                      void *            bytes,
-                                      size_t            byteCount );
-
-void              tr_peerIoReadUint8( tr_peerIo *       io,
-                                      struct evbuffer * inbuf,
-                                      uint8_t *         setme );
-
-void              tr_peerIoReadUint16( tr_peerIo *       io,
-                                       struct evbuffer * inbuf,
-                                       uint16_t *        setme );
-
-void              tr_peerIoReadUint32( tr_peerIo *       io,
-                                       struct evbuffer * inbuf,
-                                       uint32_t *        setme );
-
-void              tr_peerIoDrain( tr_peerIo *       io,
-                                  struct evbuffer * inbuf,
-                                  size_t            byteCount );
-
-/**
-***
-**/
-
-size_t            tr_peerIoGetWriteBufferSpace( const tr_peerIo * io );
-
-void              tr_peerIoSetBandwidth( tr_peerIo            * io,
-                                         struct tr_bandwidth  * bandwidth );
-
-void              tr_peerIoBandwidthUsed( tr_peerIo           * io,
-                                          tr_direction          direction,
-                                          size_t                byteCount,
-                                          int                   isPieceData );
+void      tr_peerIoSetEncryption( tr_peerIo * io,
+                                  int         encryptionMode );
+
+int       tr_peerIoIsEncrypted( const tr_peerIo * io );
+
+void      tr_peerIoWriteBytes( tr_peerIo *       io,
+                               struct evbuffer * outbuf,
+                               const void *      bytes,
+                               size_t            byteCount );
+
+void      tr_peerIoWriteUint8( tr_peerIo *       io,
+                               struct evbuffer * outbuf,
+                               uint8_t           writeme );
+
+void      tr_peerIoWriteUint16( tr_peerIo *       io,
+                                struct evbuffer * outbuf,
+                                uint16_t          writeme );
+
+void      tr_peerIoWriteUint32( tr_peerIo *       io,
+                                struct evbuffer * outbuf,
+                                uint32_t          writeme );
+
+void      tr_peerIoReadBytes( tr_peerIo *       io,
+                              struct evbuffer * inbuf,
+                              void *            bytes,
+                              size_t            byteCount );
+
+void      tr_peerIoReadUint8( tr_peerIo *       io,
+                              struct evbuffer * inbuf,
+                              uint8_t *         setme );
+
+void      tr_peerIoReadUint16( tr_peerIo *       io,
+                               struct evbuffer * inbuf,
+                               uint16_t *        setme );
+
+void      tr_peerIoReadUint32( tr_peerIo *       io,
+                               struct evbuffer * inbuf,
+                               uint32_t *        setme );
+
+void      tr_peerIoDrain( tr_peerIo *       io,
+                          struct evbuffer * inbuf,
+                          size_t            byteCount );
+
+/**
+***
+**/
+
+size_t    tr_peerIoGetWriteBufferSpace( const tr_peerIo * io );
+
+void      tr_peerIoSetBandwidth( tr_peerIo            * io,
+                                 struct tr_bandwidth  * bandwidth );
+
+void      tr_peerIoBandwidthUsed( tr_peerIo           * io,
+                                  tr_direction          direction,
+                                  size_t                byteCount,
+                                  int                   isPieceData );
+
+/**
+***
+**/
+
+tr_bool   tr_peerIoHasBandwidthLeft( const tr_peerIo  * io,
+                                     tr_direction       direction );
+
+void      tr_peerIoSetEnabled( tr_peerIo    * io,
+                               tr_direction   dir,
+                               tr_bool        isEnabled );
+                       
+int       tr_peerIoFlush( tr_peerIo     * io,
+                          tr_direction    dir,
+                          size_t          byteLimit );
+
+struct evbuffer * tr_peerIoGetReadBuffer( tr_peerIo * io );
+
 
 

branches/1.4x/libtransmission/peer-mgr-private.h

@@ -53 +53 @@
     uint8_t                  encryption_preference;
     uint16_t                 port;
-    struct in_addr           in_addr;
+    struct in_addr           addr;
     struct tr_peerIo       * io;
 

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

@@ -58 +58 @@
    
     /* how frequently to reallocate bandwidth */
-    BANDWIDTH_PERIOD_MSEC = 250,
+    BANDWIDTH_PERIOD_MSEC = 500,
 
     /* max # of peers to ask fer per torrent per reconnect pulse */
@@ -90 +90 @@
 **/
 
-/* We keep one of these for every peer we know about, whether
- * it's connected or not, so the struct must be small.
- * When our current connections underperform, we dip back
- * into this list for new ones. */
+enum
+{
+    UPLOAD_ONLY_UKNOWN,
+    UPLOAD_ONLY_YES,
+    UPLOAD_ONLY_NO
+};
+
+/**
+ * Peer information that should be kept even before we've connected and
+ * after we've disconnected.  These are kept in a pool of peer_atoms to decide
+ * which ones would make good candidates for connecting to, and to watch out
+ * for banned peers.
+ *
+ * @see tr_peer
+ * @see tr_peermsgs
+ */
 struct peer_atom
 {
-    uint8_t           from;
-    uint8_t           flags; /* these match the added_f flags */
-    uint8_t           myflags; /* flags that aren't defined in added_f */
-    uint16_t          port;
-    uint16_t          numFails;
-    struct in_addr    addr;
-    time_t            time; /* when the peer's connection status last changed */
-    time_t            piece_data_time;
+    uint8_t     from;
+    uint8_t     flags;       /* these match the added_f flags */
+    uint8_t     myflags;     /* flags that aren't defined in added_f */
+    uint8_t     uploadOnly;  /* UPLOAD_ONLY_ */
+    tr_port     port;
+    uint16_t    numFails;
+    tr_address  addr;
+    time_t      time;        /* when the peer's connection status last changed */
+    time_t      piece_data_time;
 };
 
@@ -131 +144 @@
     tr_ptrArray     * torrents; /* Torrent */
     tr_ptrArray     * incomingHandshakes; /* tr_handshake */
+    tr_ptrArray     * finishedHandshakes; /* tr_handshake */
     tr_timer        * bandwidthTimer;
 };
@@ -185 +199 @@
 
 static int
-compareAddresses( const struct in_addr * a,
-                  const struct in_addr * b )
-{
-    if( a->s_addr != b->s_addr )
-        return a->s_addr < b->s_addr ? -1 : 1;
-
-    return 0;
-}
-
-static int
-handshakeCompareToAddr( const void * va,
-                        const void * vb )
+handshakeCompareToAddr( const void * va, const void * vb )
 {
     const tr_handshake * a = va;
 
-    return compareAddresses( tr_handshakeGetAddr( a, NULL ), vb );
-}
-
-static int
-handshakeCompare( const void * a,
-                  const void * b )
+    return tr_compareAddresses( tr_handshakeGetAddr( a, NULL ), vb );
+}
+
+static int
+handshakeCompare( const void * a, const void * b )
 {
     return handshakeCompareToAddr( a, tr_handshakeGetAddr( b, NULL ) );
@@ -211 +213 @@
 
 static tr_handshake*
-getExistingHandshake( tr_ptrArray *          handshakes,
-                      const struct in_addr * in_addr )
-{
-    return tr_ptrArrayFindSorted( handshakes,
-                                  in_addr,
-                                  handshakeCompareToAddr );
-}
-
-static int
-comparePeerAtomToAddress( const void * va,
-                          const void * vb )
+getExistingHandshake( tr_ptrArray      * handshakes,
+                      const tr_address * addr )
+{
+    return tr_ptrArrayFindSorted( handshakes, addr, handshakeCompareToAddr );
+}
+
+static int
+comparePeerAtomToAddress( const void * va, const void * vb )
 {
     const struct peer_atom * a = va;
 
-    return compareAddresses( &a->addr, vb );
-}
-
-static int
-comparePeerAtoms( const void * va,
-                  const void * vb )
+    return tr_compareAddresses( &a->addr, vb );
+}
+
+static int
+comparePeerAtoms( const void * va, const void * vb )
 {
     const struct peer_atom * b = vb;
@@ -271 +269 @@
 
 static int
-peerCompare( const void * va,
-             const void * vb )
+peerCompare( const void * va, const void * vb )
 {
     const tr_peer * a = va;
     const tr_peer * b = vb;
 
-    return compareAddresses( &a->in_addr, &b->in_addr );
-}
-
-static int
-peerCompareToAddr( const void * va,
-                   const void * vb )
+    return tr_compareAddresses( &a->addr, &b->addr );
+}
+
+static int
+peerCompareToAddr( const void * va, const void * vb )
 {
     const tr_peer * a = va;
 
-    return compareAddresses( &a->in_addr, vb );
+    return tr_compareAddresses( &a->addr, vb );
 }
 
 static tr_peer*
-getExistingPeer( Torrent *              torrent,
-                 const struct in_addr * in_addr )
+getExistingPeer( Torrent          * torrent,
+                 const tr_address * addr )
 {
     assert( torrentIsLocked( torrent ) );
-    assert( in_addr );
-
-    return tr_ptrArrayFindSorted( torrent->peers,
-                                  in_addr,
-                                  peerCompareToAddr );
+    assert( addr );
+
+    return tr_ptrArrayFindSorted( torrent->peers, addr, peerCompareToAddr );
 }
 
 static struct peer_atom*
-getExistingAtom( const                  Torrent * t,
-                 const struct in_addr * addr )
+getExistingAtom( const Torrent    * t,
+                 const tr_address * addr )
 {
     assert( torrentIsLocked( t ) );
@@ -309 +303 @@
 }
 
-static int
-peerIsInUse( const Torrent *        ct,
-             const struct in_addr * addr )
+static tr_bool
+peerIsInUse( const Torrent    * ct,
+             const tr_address * addr )
 {
     Torrent * t = (Torrent*) ct;
@@ -318 +312 @@
 
     return getExistingPeer( t, addr )
-           || getExistingHandshake( t->outgoingHandshakes, addr )
-           || getExistingHandshake( t->manager->incomingHandshakes, addr );
+        || getExistingHandshake( t->outgoingHandshakes, addr )
+        || getExistingHandshake( t->manager->incomingHandshakes, addr );
 }
 
 static tr_peer*
-peerConstructor( tr_torrent * tor, const struct in_addr * in_addr )
+peerConstructor( tr_torrent * tor, const tr_address * addr )
 {
     tr_peer * p;
-
     p = tr_new0( tr_peer, 1 );
-    memcpy( &p->in_addr, in_addr, sizeof( struct in_addr ) );
+    p->addr = *addr;
     p->bandwidth = tr_bandwidthNew( tor->session, tor->bandwidth );
     return p;
@@ -334 +327 @@
 
 static tr_peer*
-getPeer( Torrent *              torrent,
-         const struct in_addr * in_addr )
+getPeer( Torrent          * torrent,
+         const tr_address * addr )
 {
     tr_peer * peer;
@@ -341 +334 @@
     assert( torrentIsLocked( torrent ) );
 
-    peer = getExistingPeer( torrent, in_addr );
+    peer = getExistingPeer( torrent, addr );
 
     if( peer == NULL )
     {
-        peer = peerConstructor( torrent->tor, in_addr );
+        peer = peerConstructor( torrent->tor, addr );
         tr_ptrArrayInsertSorted( torrent->peers, peer, peerCompare );
     }
@@ -356 +349 @@
 {
     assert( peer );
-    assert( peer->msgs );
-
-    tr_peerMsgsUnsubscribe( peer->msgs, peer->msgsTag );
-    tr_peerMsgsFree( peer->msgs );
+
+    if( peer->msgs != NULL )
+    {
+        tr_peerMsgsUnsubscribe( peer->msgs, peer->msgsTag );
+        tr_peerMsgsFree( peer->msgs );
+    }
 
     tr_peerIoFree( peer->io );
@@ -381 +376 @@
     assert( torrentIsLocked( t ) );
 
-    atom = getExistingAtom( t, &peer->in_addr );
+    atom = getExistingAtom( t, &peer->addr );
     assert( atom );
     atom->time = time( NULL );
@@ -456 +451 @@
 }
 
-/**
- * For explanation, see http://www.bittorrent.org/fast_extensions.html
- * Also see the "test-allowed-set" unit test
- *
- * @param k number of pieces in set
- * @param sz number of pieces in the torrent
- * @param infohash torrent's SHA1 hash
- * @param ip peer's address
- */
-struct tr_bitfield *
-tr_peerMgrGenerateAllowedSet(
-    const uint32_t         k,
-    const uint32_t         sz,
-    const                  uint8_t        *
-                           infohash,
-    const struct in_addr * ip )
-{
-    uint8_t       w[SHA_DIGEST_LENGTH + 4];
-    uint8_t       x[SHA_DIGEST_LENGTH];
-    tr_bitfield * a;
-    uint32_t      a_size;
-
-    *(uint32_t*)w = ntohl( htonl( ip->s_addr ) & 0xffffff00 );   /* (1) */
-    memcpy( w + 4, infohash, SHA_DIGEST_LENGTH );              /* (2) */
-    tr_sha1( x, w, sizeof( w ), NULL );                        /* (3) */
-
-    a = tr_bitfieldNew( sz );
-    a_size = 0;
-
-    while( a_size < k )
-    {
-        int i;
-        for( i = 0; i < 5 && a_size < k; ++i )                      /* (4) */
-        {
-            uint32_t j = i * 4;                                /* (5) */
-            uint32_t y = ntohl( *( uint32_t* )( x + j ) );             /* (6) */
-            uint32_t index = y % sz;                           /* (7) */
-            if( !tr_bitfieldHas( a, index ) )                  /* (8) */
-            {
-                tr_bitfieldAdd( a, index );                    /* (9) */
-                ++a_size;
-            }
-        }
-        tr_sha1( x, x, sizeof( x ), NULL );                    /* (3) */
-    }
-
-    return a;
-}
 
 static int bandwidthPulse( void * vmgr );
@@ -516 +463 @@
     m->torrents = tr_ptrArrayNew( );
     m->incomingHandshakes = tr_ptrArrayNew( );
+    m->finishedHandshakes = tr_ptrArrayNew( );
     m->bandwidthTimer = tr_timerNew( session, bandwidthPulse, m, BANDWIDTH_PERIOD_MSEC );
     return m;
@@ -523 +471 @@
 tr_peerMgrFree( tr_peerMgr * manager )
 {
+    tr_handshake * handshake;
+
     managerLock( manager );
 
@@ -534 +484 @@
     tr_ptrArrayFree( manager->incomingHandshakes, NULL );
 
+    while(( handshake = tr_ptrArrayPop( manager->finishedHandshakes )))
+        tr_handshakeFree( handshake );
+
+    tr_ptrArrayFree( manager->finishedHandshakes, NULL );
+
     /* free the torrents. */
     tr_ptrArrayFree( manager->torrents, torrentDestructor );
@@ -542 +497 @@
 
 static tr_peer**
-getConnectedPeers( Torrent * t,
-                   int *     setmeCount )
-{
-    int       i, peerCount, connectionCount;
+getConnectedPeers( Torrent * t, int * setmeCount )
+{
+    int i, peerCount, connectionCount;
     tr_peer **peers;
     tr_peer **ret;
@@ -578 +532 @@
 ***/
 
-int
-tr_peerMgrPeerIsSeed( const tr_peerMgr *     mgr,
-                      const uint8_t *        torrentHash,
-                      const struct in_addr * addr )
-{
-    int                      isSeed = FALSE;
-    const Torrent *          t = NULL;
+tr_bool
+tr_peerMgrPeerIsSeed( const tr_peerMgr  * mgr,
+                      const uint8_t     * torrentHash,
+                      const tr_address  * addr )
+{
+    tr_bool isSeed = FALSE;
+    const Torrent * t = NULL;
     const struct peer_atom * atom = NULL;
 
@@ -648 +602 @@
 
 static int
-compareRefillPiece( const void * aIn,
-                    const void * bIn )
+compareRefillPiece( const void * aIn, const void * bIn )
 {
     const struct tr_refill_piece * a = aIn;
@@ -678 +631 @@
 
 static tr_piece_index_t *
-getPreferredPieces( Torrent           * t,
-                    tr_piece_index_t  * pieceCount )
+getPreferredPieces( Torrent * t, tr_piece_index_t * pieceCount )
 {
     const tr_torrent  * tor = t->tor;
@@ -877 +829 @@
         for( j=0; !handled && j<peerCount; )
         {
-            const int val = tr_peerMsgsAddRequest( peers[j]->msgs,
-                                                   index, offset, length );
+            const tr_addreq_t val = tr_peerMsgsAddRequest( peers[j]->msgs, index, offset, length );
             switch( val )
             {
@@ -905 +856 @@
         for( j=0; !handled && j<webseedCount; )
         {
-            const tr_addreq_t val = tr_webseedAddRequest( webseeds[j],
-                                                          index, offset, length );
+            const tr_addreq_t val = tr_webseedAddRequest( webseeds[j], index, offset, length );
             switch( val )
             {
@@ -943 +893 @@
     assert( torrentIsLocked( t ) );
 
+    tordbg( t, "got a block; cancelling any duplicate requests from peers %"PRIu32":%"PRIu32"->%"PRIu32, index, offset, length );
     peers = getConnectedPeers( t, &size );
     for( i=0; i<size; ++i )
@@ -954 +905 @@
 {
     tordbg( t, "increasing peer %s strike count to %d",
-            tr_peerIoAddrStr( &peer->in_addr,
+            tr_peerIoAddrStr( &peer->addr,
                               peer->port ), peer->strikes + 1 );
 
     if( ++peer->strikes >= MAX_BAD_PIECES_PER_PEER )
     {
-        struct peer_atom * atom = getExistingAtom( t, &peer->in_addr );
+        struct peer_atom * atom = getExistingAtom( t, &peer->addr );
         atom->myflags |= MYFLAG_BANNED;
         peer->doPurge = 1;
@@ -988 +939 @@
 
 static void
-peerCallbackFunc( void * vpeer,
-                  void * vevent,
-                  void * vt )
-{
-    tr_peer *             peer = vpeer; /* may be NULL if peer is a webseed */
-    Torrent *             t = (Torrent *) vt;
+peerSuggestedPiece( Torrent            * t UNUSED,
+                    tr_peer            * peer UNUSED,
+                    tr_piece_index_t     pieceIndex UNUSED,
+                    int                  isFastAllowed UNUSED )
+{
+#if 0
+    assert( t );
+    assert( peer );
+    assert( peer->msgs );
+
+    /* is this a valid piece? */
+    if(  pieceIndex >= t->tor->info.pieceCount )
+        return;
+
+    /* don't ask for it if we've already got it */
+    if( tr_cpPieceIsComplete( t->tor->completion, pieceIndex ) )
+        return;
+
+    /* don't ask for it if they don't have it */
+    if( !tr_bitfieldHas( peer->have, pieceIndex ) )
+        return;
+
+    /* don't ask for it if we're choked and it's not fast */
+    if( !isFastAllowed && peer->clientIsChoked )
+        return;
+
+    /* request the blocks that we don't have in this piece */
+    {
+        tr_block_index_t block;
+        const tr_torrent * tor = t->tor;
+        const tr_block_index_t start = tr_torPieceFirstBlock( tor, pieceIndex );
+        const tr_block_index_t end = start + tr_torPieceCountBlocks( tor, pieceIndex );
+
+        for( block=start; block<end; ++block )
+        {
+            if( !tr_cpBlockIsComplete( tor->completion, block ) )
+            {
+                const uint32_t offset = getBlockOffsetInPiece( tor, block );
+                const uint32_t length = tr_torBlockCountBytes( tor, block );
+                tr_peerMsgsAddRequest( peer->msgs, pieceIndex, offset, length );
+                incrementPieceRequests( t, pieceIndex );
+            }
+        }
+    }
+#endif
+}
+
+static void
+peerCallbackFunc( void * vpeer, void * vevent, void * vt )
+{
+    tr_peer * peer = vpeer; /* may be NULL if peer is a webseed */
+    Torrent * t = vt;
     const tr_peer_event * e = vevent;
 
@@ -1000 +997 @@
     switch( e->eventType )
     {
+        case TR_PEER_UPLOAD_ONLY:
+            /* update our atom */
+            if( peer ) {
+                struct peer_atom * a = getExistingAtom( t, &peer->addr );
+                a->uploadOnly = e->uploadOnly ? UPLOAD_ONLY_YES : UPLOAD_ONLY_NO;
+            }
+            break;
+
         case TR_PEER_NEED_REQ:
             refillSoon( t );
@@ -1024 +1029 @@
             /* update our atom */
             if( peer ) {
-                struct peer_atom * a = getExistingAtom( t, &peer->in_addr );
+                struct peer_atom * a = getExistingAtom( t, &peer->addr );
                 if( e->wasPieceData )
                     a->piece_data_time = now;
@@ -1031 +1036 @@
             break;
         }
+
+        case TR_PEER_CLIENT_GOT_SUGGEST:
+            if( peer )
+                peerSuggestedPiece( t, peer, e->pieceIndex, FALSE );
+            break;
+
+        case TR_PEER_CLIENT_GOT_ALLOWED_FAST:
+            if( peer )
+                peerSuggestedPiece( t, peer, e->pieceIndex, TRUE );
+            break;
 
         case TR_PEER_CLIENT_GOT_DATA:
@@ -1036 +1051 @@
             const time_t now = time( NULL );
             tr_torrent * tor = t->tor;
-
             tor->activityDate = now;
 
@@ -1054 +1068 @@
             /* update our atom */
             if( peer ) {
-                struct peer_atom * a = getExistingAtom( t, &peer->in_addr );
+                struct peer_atom * a = getExistingAtom( t, &peer->addr );
                 if( e->wasPieceData )
                     a->piece_data_time = now;
@@ -1066 +1080 @@
             if( peer )
             {
-                struct peer_atom * atom = getExistingAtom( t,
-                                                           &peer->in_addr );
-                const int          peerIsSeed = e->progress >= 1.0;
-                if( peerIsSeed )
-                {
-                    tordbg( t, "marking peer %s as a seed",
-                           tr_peerIoAddrStr( &atom->addr,
-                                             atom->port ) );
+                struct peer_atom * atom = getExistingAtom( t, &peer->addr );
+                const int peerIsSeed = e->progress >= 1.0;
+                if( peerIsSeed ) {
+                    tordbg( t, "marking peer %s as a seed", tr_peerIoAddrStr( &atom->addr, atom->port ) );
                     atom->flags |= ADDED_F_SEED_FLAG;
-                }
-                else
-                {
-                    tordbg( t, "marking peer %s as a non-seed",
-                           tr_peerIoAddrStr( &atom->addr,
-                                             atom->port ) );
+                } else {
+                    tordbg( t, "marking peer %s as a non-seed", tr_peerIoAddrStr( &atom->addr, atom->port ) );
                     atom->flags &= ~ADDED_F_SEED_FLAG;
                 }
@@ -1091 +1097 @@
             tr_torrent *     tor = t->tor;
 
-            tr_block_index_t block = _tr_block( tor, e->pieceIndex,
-                                                e->offset );
+            tr_block_index_t block = _tr_block( tor, e->pieceIndex, e->offset );
 
             tr_cpBlockAdd( tor->completion, block );
@@ -1102 +1107 @@
             {
                 const tr_piece_index_t p = e->pieceIndex;
-                const int              ok = tr_ioTestPiece( tor, p );
+                const tr_bool ok = tr_ioTestPiece( tor, p );
 
                 if( !ok )
                 {
-                    tr_torerr( tor,
-                              _( "Piece %lu, which was just downloaded, failed its checksum test" ),
-                              (unsigned long)p );
+                    tr_torerr( tor, _( "Piece %lu, which was just downloaded, failed its checksum test" ),
+                               (unsigned long)p );
                 }
 
@@ -1119 +1123 @@
                 else
                 {
-                    int        i, peerCount;
+                    int i, peerCount;
                     tr_peer ** peers = getConnectedPeers( t, &peerCount );
                     for( i = 0; i < peerCount; ++i )
@@ -1136 +1140 @@
                 addStrike( t, peer );
                 peer->doPurge = 1;
+                tordbg( t, "setting doPurge because we got an EINVAL error" );
             }
             else if( ( e->err == ERANGE )
@@ -1143 +1148 @@
                 /* 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 */
@@ -1162 +1168 @@
 
 static void
-ensureAtomExists( Torrent *              t,
-                  const struct in_addr * addr,
-                  uint16_t               port,
-                  uint8_t                flags,
-                  uint8_t                from )
+ensureAtomExists( Torrent          * t,
+                  const tr_address * addr,
+                  tr_port            port,
+                  uint8_t            flags,
+                  uint8_t            from )
 {
     if( getExistingAtom( t, addr ) == NULL )
@@ -1176 +1182 @@
         a->flags = flags;
         a->from = from;
-        tordbg( t, "got a new atom: %s",
-               tr_peerIoAddrStr( &a->addr, a->port ) );
+        tordbg( t, "got a new atom: %s", tr_peerIoAddrStr( &a->addr, a->port ) );
         tr_ptrArrayInsertSorted( t->pool, a, comparePeerAtoms );
     }
@@ -1191 +1196 @@
 getPeerCount( const Torrent * t )
 {
-    return tr_ptrArraySize( t->peers ) + tr_ptrArraySize(
-               t->outgoingHandshakes );
+    return tr_ptrArraySize( t->peers ) + tr_ptrArraySize( t->outgoingHandshakes );
 }
 
 /* FIXME: this is kind of a mess. */
-static int
-myHandshakeDoneCB( tr_handshake *  handshake,
-                   tr_peerIo *     io,
+static tr_bool
+myHandshakeDoneCB( tr_handshake  * handshake,
+                   tr_peerIo     * io,
                    int             isConnected,
                    const uint8_t * peer_id,
-                   void *          vmanager )
-{
-    int                    ok = isConnected;
-    int                    success = FALSE;
-    uint16_t               port;
-    const struct in_addr * addr;
-    tr_peerMgr *           manager = (tr_peerMgr*) vmanager;
-    Torrent *              t;
-    tr_handshake *         ours;
+                   void          * vmanager )
+{
+    tr_bool            ok = isConnected;
+    tr_bool            success = FALSE;
+    tr_port            port;
+    const tr_address * addr;
+    tr_peerMgr       * manager = vmanager;
+    Torrent          * t;
+    tr_handshake     * ours;
 
     assert( io );
@@ -1243 +1247 @@
                 ++atom->numFails;
         }
-
-        tr_peerIoFree( io );
     }
     else /* looking good */
@@ -1257 +1259 @@
         {
             tordbg( t, "banned peer %s tried to reconnect",
-                   tr_peerIoAddrStr( &atom->addr,
-                                     atom->port ) );
-            tr_peerIoFree( io );
+                    tr_peerIoAddrStr( &atom->addr, atom->port ) );
         }
         else if( tr_peerIoIsIncoming( io )
@@ -1265 +1265 @@
 
         {
-            tr_peerIoFree( io );
         }
         else
@@ -1273 +1272 @@
             if( peer ) /* we already have this peer */
             {
-                tr_peerIoFree( io );
             }
             else
@@ -1289 +1287 @@
 
                 peer->port = port;
-                peer->io = io;
-                peer->msgs = tr_peerMsgsNew( t->tor, peer, peerCallbackFunc, t, &peer->msgsTag );
+                peer->io = tr_handshakeStealIO( handshake );
+                tr_peerMsgsNew( t->tor, peer, peerCallbackFunc, t, &peer->msgsTag );
                 tr_peerIoSetBandwidth( io, peer->bandwidth );
 
@@ -1298 +1296 @@
     }
 
+    if( !success )
+        tr_ptrArrayAppend( manager->finishedHandshakes, handshake );
+
     if( t )
         torrentUnlock( t );
@@ -1305 +1306 @@
 
 void
-tr_peerMgrAddIncoming( tr_peerMgr *     manager,
-                       struct in_addr * addr,
-                       uint16_t         port,
-                       int              socket )
+tr_peerMgrAddIncoming( tr_peerMgr * manager,
+                       tr_address * addr,
+                       tr_port      port,
+                       int          socket )
 {
     managerLock( manager );
@@ -1314 +1315 @@
     if( tr_sessionIsAddressBlocked( manager->session, addr ) )
     {
-        tr_dbg( "Banned IP address \"%s\" tried to connect to us",
-               inet_ntoa( *addr ) );
+        tr_dbg( "Banned IP address \"%s\" tried to connect to us", inet_ntoa( *addr ) );
         tr_netClose( socket );
     }
@@ -1339 +1339 @@
 
     managerUnlock( manager );
+}
+
+static tr_bool
+tr_isPex( const tr_pex * pex )
+{
+    return pex && tr_isAddress( &pex->addr );
 }
 
@@ -1347 +1353 @@
                   const tr_pex *  pex )
 {
-    Torrent * t;
-
-    managerLock( manager );
-
-    t = getExistingTorrent( manager, torrentHash );
-    if( !tr_sessionIsAddressBlocked( t->manager->session, &pex->in_addr ) )
-        ensureAtomExists( t, &pex->in_addr, pex->port, pex->flags, from );
-
-    managerUnlock( manager );
+    if( tr_isPex( pex ) ) /* safeguard against corrupt data */
+    {
+        Torrent * t;
+        managerLock( manager );
+
+        t = getExistingTorrent( manager, torrentHash );
+        if( !tr_sessionIsAddressBlocked( t->manager->session, &pex->addr ) )
+            ensureAtomExists( t, &pex->addr, pex->port, pex->flags, from );
+
+        managerUnlock( manager );
+    }
 }
 
@@ -1372 +1380 @@
     for( i = 0; i < n; ++i )
     {
-        memcpy( &pex[i].in_addr, walk, 4 ); walk += 4;
+        memcpy( &pex[i].addr, walk, 4 ); walk += 4;
         memcpy( &pex[i].port, walk, 2 ); walk += 2;
         if( added_f && ( n == added_f_len ) )
@@ -1406 +1414 @@
             if( tr_bitfieldHas( peer->blame, pieceIndex ) )
             {
-                tordbg(
-                    t,
-                    "peer %s contributed to corrupt piece (%d); now has %d strikes",
-                    tr_peerIoAddrStr( &peer->in_addr, peer->port ),
-                    pieceIndex, (int)peer->strikes + 1 );
+                tordbg( t, "peer %s contributed to corrupt piece (%d); now has %d strikes",
+                        tr_peerIoAddrStr( &peer->addr, peer->port ),
+                        pieceIndex, (int)peer->strikes + 1 );
                 addStrike( t, peer );
             }
@@ -1418 +1424 @@
 
 int
-tr_pexCompare( const void * va,
-               const void * vb )
+tr_pexCompare( const void * va, const void * vb )
 {
     const tr_pex * a = va;
     const tr_pex * b = vb;
-    int            i =
-        memcmp( &a->in_addr, &b->in_addr, sizeof( struct in_addr ) );
-
-    if( i ) return i;
-    if( a->port < b->port ) return -1;
-    if( a->port > b->port ) return 1;
+    int i;
+
+    assert( tr_isPex( a ) );
+    assert( tr_isPex( b ) );
+
+    if(( i = tr_compareAddresses( &a->addr, &b->addr )))
+        return i;
+
+    if( a->port != b->port )
+        return a->port < b->port ? -1 : 1;
+
     return 0;
 }
-
-int tr_pexCompare( const void * a,
-                   const void * b );
 
 static int
@@ -1448 +1455 @@
 
 int
-tr_peerMgrGetPeers( tr_peerMgr *    manager,
-                    const uint8_t * torrentHash,
-                    tr_pex **       setme_pex )
+tr_peerMgrGetPeers( tr_peerMgr      * manager,
+                    const uint8_t   * torrentHash,
+                    tr_pex         ** setme_pex )
 {
     int peerCount = 0;
@@ -1458 +1465 @@
 
     t = getExistingTorrent( (tr_peerMgr*)manager, torrentHash );
-    if( !t )
+    if( t == NULL )
     {
         *setme_pex = NULL;
@@ -1469 +1476 @@
         tr_pex * walk = pex;
 
-        for( i = 0; i < peerCount; ++i, ++walk )
+        for( i=0; i<peerCount; ++i, ++walk )
         {
             const tr_peer * peer = peers[i];
-            walk->in_addr = peer->in_addr;
+            const struct peer_atom * atom = getExistingAtom( t, &peer->addr );
+
+            assert( tr_isAddress( &peer->addr ) );
+            walk->addr = peer->addr;
             walk->port = peer->port;
             walk->flags = 0;
-            if( peerPrefersCrypto( peer ) ) walk->flags |= ADDED_F_ENCRYPTION_FLAG;
-            if( peer->progress >= 1.0 ) walk->flags |= ADDED_F_SEED_FLAG;
+            if( peerPrefersCrypto( peer ) ) 
+                walk->flags |= ADDED_F_ENCRYPTION_FLAG;
+            if( ( atom->uploadOnly == UPLOAD_ONLY_YES ) || ( peer->progress >= 1.0 ) )
+                walk->flags |= ADDED_F_SEED_FLAG;
         }
 
@@ -1603 +1615 @@
     const tr_torrent * tor;
     float              interval;
-    int                isComplete;
+    tr_bool            isSeed;
     int                peerCount;
     const tr_peer **   peers;
@@ -1612 +1624 @@
     tor = t->tor;
     interval = tor->info.pieceCount / (float)tabCount;
-    isComplete = tor
-                 && ( tr_cpGetStatus ( tor->completion ) == TR_CP_COMPLETE );
+    isSeed = tor && ( tr_cpGetStatus ( tor->completion ) == TR_CP_COMPLETE );
     peers = (const tr_peer **) tr_ptrArrayPeek( t->peers, &peerCount );
 
@@ -1622 +1633 @@
         const int piece = i * interval;
 
-        if( isComplete || tr_cpPieceIsComplete( tor->completion, piece ) )
+        if( isSeed || tr_cpPieceIsComplete( tor->completion, piece ) )
             tab[i] = -1;
-        else if( peerCount )
-        {
+        else if( peerCount ) {
             int j;
             for( j = 0; j < peerCount; ++j )
@@ -1645 +1655 @@
     tr_peer **    peers;
     tr_bitfield * pieces;
-
     managerLock( manager );
 
@@ -1663 +1672 @@
                           const uint8_t *    torrentHash )
 {
-    int             ret;
+    int ret;
     const Torrent * t;
-
     managerLock( manager );
 
     t = getExistingTorrent( (tr_peerMgr*)manager, torrentHash );
-    ret = t && ( !tr_ptrArrayEmpty( t->peers )
-               || !tr_ptrArrayEmpty( t->webseeds ) );
+    ret = t && ( !tr_ptrArrayEmpty( t->peers ) || !tr_ptrArrayEmpty( t->webseeds ) );
 
     managerUnlock( manager );
@@ -1678 +1685 @@
 void
 tr_peerMgrTorrentStats( const tr_peerMgr * manager,
-                        const uint8_t *    torrentHash,
-                        int *              setmePeersKnown,
-                        int *              setmePeersConnected,
-                        int *              setmeSeedsConnected,
-                        int *              setmeWebseedsSendingToUs,
-                        int *              setmePeersSendingToUs,
-                        int *              setmePeersGettingFromUs,
-                        int *              setmePeersFrom )
-{
-    int                 i, size;
-    const Torrent *     t;
-    const tr_peer **    peers;
+                        const uint8_t    * torrentHash,
+                        int              * setmePeersKnown,
+                        int              * setmePeersConnected,
+                        int              * setmeSeedsConnected,
+                        int              * setmeWebseedsSendingToUs,
+                        int              * setmePeersSendingToUs,
+                        int              * setmePeersGettingFromUs,
+                        int              * setmePeersFrom )
+{
+    int i, size;
+    const Torrent * t;
+    const tr_peer ** peers;
     const tr_webseed ** webseeds;
 
@@ -1704 +1711 @@
     *setmeWebseedsSendingToUs  = 0;
 
-    for( i = 0; i < TR_PEER_FROM__MAX; ++i )
+    for( i=0; i<TR_PEER_FROM__MAX; ++i )
         setmePeersFrom[i] = 0;
 
-    for( i = 0; i < size; ++i )
-    {
-        const tr_peer *          peer = peers[i];
-        const struct peer_atom * atom = getExistingAtom( t, &peer->in_addr );
+    for( i=0; i<size; ++i )
+    {
+        const tr_peer * peer = peers[i];
+        const struct peer_atom * atom = getExistingAtom( t, &peer->addr );
 
         if( peer->io == NULL ) /* not connected */
             continue;
 
-        ++ * setmePeersConnected;
+        ++*setmePeersConnected;
 
         ++setmePeersFrom[atom->from];
 
         if( clientIsDownloadingFrom( peer ) )
-            ++ * setmePeersSendingToUs;
+            ++*setmePeersSendingToUs;
 
         if( clientIsUploadingTo( peer ) )
-            ++ * setmePeersGettingFromUs;
+            ++*setmePeersGettingFromUs;
 
         if( atom->flags & ADDED_F_SEED_FLAG )
-            ++ * setmeSeedsConnected;
+            ++*setmeSeedsConnected;
     }
 
     webseeds = (const tr_webseed **) tr_ptrArrayPeek( t->webseeds, &size );
-    for( i = 0; i < size; ++i )
-    {
+    for( i=0; i<size; ++i )
         if( tr_webseedIsActive( webseeds[i] ) )
-            ++ * setmeWebseedsSendingToUs;
-    }
+            ++*setmeWebseedsSendingToUs;
 
     managerUnlock( manager );
@@ -1743 +1748 @@
                      const uint8_t *    torrentHash )
 {
-    const Torrent *     t;
+    const Torrent * t;
     const tr_webseed ** webseeds;
-    int                 i;
-    int                 webseedCount;
-    float *             ret;
+    int i;
+    int webseedCount;
+    float * ret;
 
     assert( manager );
@@ -1758 +1763 @@
     ret = tr_new0( float, webseedCount );
 
-    for( i = 0; i < webseedCount; ++i )
+    for( i=0; i<webseedCount; ++i )
         if( !tr_webseedGetSpeed( webseeds[i], &ret[i] ) )
             ret[i] = -1.0;
@@ -1767 +1772 @@
 
 double
-tr_peerGetPieceSpeed( const tr_peer    * peer,
-                      tr_direction       direction )
+tr_peerGetPieceSpeed( const tr_peer * peer, tr_direction direction )
 {
     assert( peer );
@@ -1798 +1802 @@
         char *                   pch;
         const tr_peer *          peer = peers[i];
-        const struct peer_atom * atom = getExistingAtom( t, &peer->in_addr );
+        const struct peer_atom * atom = getExistingAtom( t, &peer->addr );
         tr_peer_stat *           stat = ret + i;
 
-        tr_netNtop( &peer->in_addr, stat->addr, sizeof( stat->addr ) );
+        tr_netNtop( &peer->addr, stat->addr, sizeof( stat->addr ) );
         tr_strlcpy( stat->client, ( peer->client ? peer->client : "" ),
                    sizeof( stat->client ) );
@@ -1817 +1821 @@
         stat->isDownloadingFrom  = clientIsDownloadingFrom( peer );
         stat->isUploadingTo      = clientIsUploadingTo( peer );
+        stat->isSeed             = ( atom->uploadOnly == UPLOAD_ONLY_YES ) || ( peer->progress >= 1.0 );
 
         pch = stat->flagStr;
@@ -1824 +1829 @@
         if( stat->isUploadingTo ) *pch++ = 'U';
         else if( stat->peerIsInterested ) *pch++ = 'u';
-        if( !stat->clientIsChoked && !stat->clientIsInterested ) *pch++ =
-                'K';
+        if( !stat->clientIsChoked && !stat->clientIsInterested ) *pch++ = 'K';
         if( !stat->peerIsChoked && !stat->peerIsInterested ) *pch++ = '?';
         if( stat->isEncrypted ) *pch++ = 'E';
@@ -1899 +1903 @@
     {
         tr_peer * peer = peers[i];
+        struct peer_atom * atom = getExistingAtom( t, &peer->addr );
+
         if( peer->progress >= 1.0 ) /* choke all seeds */
+        {
             tr_peerMsgsSetChoke( peer->msgs, TRUE );
-        else if( chokeAll )
+        }
+        else if( atom->uploadOnly == UPLOAD_ONLY_YES ) /* choke partial seeds */
+        {
             tr_peerMsgsSetChoke( peer->msgs, TRUE );
-        else {
+        }
+        else if( chokeAll ) /* choke everyone if we're not uploading */
+        {
+            tr_peerMsgsSetChoke( peer->msgs, TRUE );
+        }
+        else
+        {
             struct ChokeData * n = &choke[size++];
             n->peer         = peer;
@@ -1928 +1943 @@
      */
     unchokedInterested = 0;
-    for( i = 0; i < size && unchokedInterested < MAX_UNCHOKED_PEERS; ++i )
-    {
+    for( i=0; i<size && unchokedInterested<MAX_UNCHOKED_PEERS; ++i ) {
         choke[i].doUnchoke = 1;
         if( choke[i].isInterested )
@@ -1938 +1952 @@
     if( i < size )
     {
-        int                n;
+        int n;
         struct ChokeData * c;
-        tr_ptrArray *      randPool = tr_ptrArrayNew( );
-
-        for( ; i < size; ++i )
+        tr_ptrArray * randPool = tr_ptrArrayNew( );
+
+        for( ; i<size; ++i )
         {
             if( choke[i].isInterested )
             {
                 const tr_peer * peer = choke[i].peer;
-                int             x = 1, y;
+                int x = 1, y;
                 if( isNew( peer ) ) x *= 3;
                 if( isSame( peer ) ) x *= 3;
-                for( y = 0; y < x; ++y )
+                for( y=0; y<x; ++y )
                     tr_ptrArrayAppend( randPool, &choke[i] );
             }
         }
 
-        if( ( n = tr_ptrArraySize( randPool ) ) )
+        if(( n = tr_ptrArraySize( randPool )))
         {
-            c = tr_ptrArrayNth( randPool, tr_cryptoWeakRandInt( n ) );
+            c = tr_ptrArrayNth( randPool, tr_cryptoWeakRandInt( n ));
             c->doUnchoke = 1;
             t->optimistic = c->peer;
@@ -1965 +1979 @@
     }
 
-    for( i = 0; i < size; ++i )
+    for( i=0; i<size; ++i )
         tr_peerMsgsSetChoke( choke[i].peer->msgs, !choke[i].doUnchoke );
 
@@ -1997 +2011 @@
     const tr_torrent *       tor = t->tor;
     const time_t             now = time( NULL );
-    const struct peer_atom * atom = getExistingAtom( t, &peer->in_addr );
+    const struct peer_atom * atom = getExistingAtom( t, &peer->addr );
 
     /* if it's marked for purging, close it */
@@ -2003 +2017 @@
     {
         tordbg( t, "purging peer %s because its doPurge flag is set",
-               tr_peerIoAddrStr( &atom->addr,
-                                 atom->port ) );
+                tr_peerIoAddrStr( &atom->addr, atom->port ) );
         return TRUE;
     }
@@ -2017 +2030 @@
         else if( peer->progress < tr_cpPercentDone( tor->completion ) )
             peerHasEverything = FALSE;
-        else
-        {
-            tr_bitfield * tmp =
-                tr_bitfieldDup( tr_cpPieceBitfield( tor->completion ) );
+        else {
+            tr_bitfield * tmp = tr_bitfieldDup( tr_cpPieceBitfield( tor->completion ) );
             tr_bitfieldDifference( tmp, peer->have );
             peerHasEverything = tr_bitfieldCountTrueBits( tmp ) == 0;
             tr_bitfieldFree( tmp );
         }
-        if( peerHasEverything
-          && ( !tr_torrentAllowsPex( tor ) || ( now - atom->time >= 30 ) ) )
+
+        if( peerHasEverything && ( !tr_torrentAllowsPex(tor) || (now-atom->time>=30 )))
         {
             tordbg( t, "purging peer %s because we're both seeds",
-                   tr_peerIoAddrStr( &atom->addr,
-                                     atom->port ) );
+                    tr_peerIoAddrStr( &atom->addr, atom->port ) );
             return TRUE;
         }
@@ -2060 +2070 @@
 
 static tr_peer **
-getPeersToClose( Torrent * t,
-                 int *     setmeSize )
-{
-    int               i, peerCount, outsize;
-    tr_peer **        peers = (tr_peer**) tr_ptrArrayPeek( t->peers,
-                                                           &peerCount );
+getPeersToClose( Torrent * t, int * setmeSize )
+{
+    int i, peerCount, outsize;
+    tr_peer ** peers = (tr_peer**) tr_ptrArrayPeek( t->peers, &peerCount );
     struct tr_peer ** ret = tr_new( tr_peer *, peerCount );
 
@@ -2101 +2109 @@
         return a->time < b->time ? -1 : 1;
 
+    /* all other things being equal, prefer peers whose
+     * information comes from a more reliable source */
+    if( a->from != b->from )
+        return a->from < b->from ? -1 : 1;
+
     return 0;
 }
@@ -2113 +2126 @@
      * data, try to reconnect to them sooner rather that later -- we don't
      * want network troubles to get in the way of a good peer. */
-    if( ( now - atom->piece_data_time ) <=
-       ( MINIMUM_RECONNECT_INTERVAL_SECS * 2 ) )
+    if( ( now - atom->piece_data_time ) <= ( MINIMUM_RECONNECT_INTERVAL_SECS * 2 ) )
         sec = MINIMUM_RECONNECT_INTERVAL_SECS;
 
@@ -2123 +2135 @@
     /* otherwise, the interval depends on how many times we've tried
      * and failed to connect to the peer */
-    else switch( atom->numFails )
-        {
-            case 0:
-                sec = 0; break;
-
-            case 1:
-                sec = 5; break;
-
-            case 2:
-                sec = 2 * 60; break;
-
-            case 3:
-                sec = 15 * 60; break;
-
-            case 4:
-                sec = 30 * 60; break;
-
-            case 5:
-                sec = 60 * 60; break;
-
-            default:
-                sec = 120 * 60; break;
-        }
+    else switch( atom->numFails ) {
+        case 0: sec = 0; break;
+        case 1: sec = 5; break;
+        case 2: sec = 2 * 60; break;
+        case 3: sec = 15 * 60; break;
+        case 4: sec = 30 * 60; break;
+        case 5: sec = 60 * 60; break;
+        default: sec = 120 * 60; break;
+    }
 
     return sec;
@@ -2151 +2149 @@
 
 static struct peer_atom **
-getPeerCandidates(                               Torrent * t,
-                                           int * setmeSize )
+getPeerCandidates( Torrent * t, int * setmeSize )
 {
     int                 i, atomCount, retCount;
@@ -2185 +2182 @@
 
         /* no need to connect if we're both seeds... */
-        if( seed && ( atom->flags & ADDED_F_SEED_FLAG ) )
+        if( seed && ( ( atom->flags & ADDED_F_SEED_FLAG ) ||
+                      ( atom->uploadOnly == UPLOAD_ONLY_YES ) ) )
             continue;
 
@@ -2192 +2190 @@
         if( ( now - atom->time ) < interval )
         {
-            tordbg(
-                t,
-                "RECONNECT peer %d (%s) is in its grace period of %d seconds..",
-                i, tr_peerIoAddrStr( &atom->addr,
-                                     atom->port ), interval );
+            tordbg( t, "RECONNECT peer %d (%s) is in its grace period of %d seconds..",
+                    i, tr_peerIoAddrStr( &atom->addr, atom->port ), interval );
             continue;
         }
@@ -2235 +2230 @@
     else
     {
-        int                 i, nCandidates, nBad;
+        int i, nCandidates, nBad;
         struct peer_atom ** candidates = getPeerCandidates( t, &nCandidates );
-        struct tr_peer **   connections = getPeersToClose( t, &nBad );
+        struct tr_peer ** connections = getPeersToClose( t, &nBad );
 
         if( nBad || nCandidates )
-            tordbg(
-                t, "reconnect pulse for [%s]: %d bad connections, "
-                   "%d connection candidates, %d atoms, max per pulse is %d",
-                t->tor->info.name, nBad, nCandidates,
-                tr_ptrArraySize( t->pool ),
-                (int)MAX_RECONNECTIONS_PER_PULSE );
+            tordbg( t, "reconnect pulse for [%s]: %d bad connections, "
+                    "%d connection candidates, %d atoms, max per pulse is %d",
+                    t->tor->info.name, nBad, nCandidates,
+                    tr_ptrArraySize( t->pool ),
+                    (int)MAX_RECONNECTIONS_PER_PULSE );
 
         /* disconnect some peers.
@@ -2251 +2245 @@
            so reset their `numFails' weight to zero.  otherwise we connected
            to them fruitlessly, so mark it as another fail */
-        for( i = 0; i < nBad; ++i )
-        {
-            tr_peer *          peer = connections[i];
-            struct peer_atom * atom = getExistingAtom( t, &peer->in_addr );
+        for( i = 0; i < nBad; ++i ) {
+            tr_peer * peer = connections[i];
+            struct peer_atom * atom = getExistingAtom( t, &peer->addr );
             if( atom->piece_data_time )
                 atom->numFails = 0;
             else
                 ++atom->numFails;
-            tordbg( t, "removing bad peer %s",
-                   tr_peerIoGetAddrStr( peer->io ) );
+            tordbg( t, "removing bad peer %s", tr_peerIoGetAddrStr( peer->io ) );
             removePeer( t, peer );
         }
@@ -2268 +2260 @@
            && ( i < MAX_RECONNECTIONS_PER_PULSE )
            && ( getPeerCount( t ) < getMaxPeerCount( t->tor ) )
-           && ( newConnectionsThisSecond < MAX_CONNECTIONS_PER_SECOND );
-             ++i )
+           && ( newConnectionsThisSecond < MAX_CONNECTIONS_PER_SECOND ); ++i )
         {
             tr_peerMgr *       mgr = t->manager;
@@ -2278 +2269 @@
                    tr_peerIoAddrStr( &atom->addr, atom->port ) );
 
-            io =
-                tr_peerIoNewOutgoing( mgr->session, &atom->addr, atom->port,
-                                      t->hash );
+            io = tr_peerIoNewOutgoing( mgr->session, &atom->addr, atom->port, t->hash );
             if( io == NULL )
             {
@@ -2287 +2276 @@
             else
             {
-                tr_handshake * handshake = tr_handshakeNew(
-                    io,
-                    mgr->session->
-                    encryptionMode,
-                    myHandshakeDoneCB,
-                    mgr );
+                tr_handshake * handshake = tr_handshakeNew( io,
+                                                            mgr->session->encryptionMode,
+                                                            myHandshakeDoneCB,
+                                                            mgr );
 
                 assert( tr_peerIoGetTorrentHash( io ) );
@@ -2340 +2327 @@
 bandwidthPulse( void * vmgr )
 {
+    tr_handshake * handshake;
     tr_peerMgr * mgr = vmgr;
     managerLock( mgr );
 
+    /* FIXME: this next line probably isn't necessary... */
     pumpAllPeers( mgr );
+
+    /* allocate bandwidth to the peers */
     tr_bandwidthAllocate( mgr->session->bandwidth, TR_UP, BANDWIDTH_PERIOD_MSEC );
     tr_bandwidthAllocate( mgr->session->bandwidth, TR_DOWN, BANDWIDTH_PERIOD_MSEC );
-    pumpAllPeers( mgr );
+
+    /* free all the finished handshakes */
+    while(( handshake = tr_ptrArrayPop( mgr->finishedHandshakes )))
+        tr_handshakeFree( handshake );
 
     managerUnlock( mgr );

branches/1.4x/libtransmission/peer-mgr.h

@@ -43 +43 @@
 typedef struct tr_pex
 {
-    struct in_addr in_addr;
+    tr_address  addr;
     uint16_t    port;
     uint8_t     flags;
@@ -55 +55 @@
 void tr_peerMgrFree( tr_peerMgr * manager );
 
-int tr_peerMgrPeerIsSeed( const tr_peerMgr      * mgr,
-                          const uint8_t         * torrentHash,
-                          const struct in_addr  * addr );
+tr_bool tr_peerMgrPeerIsSeed( const tr_peerMgr      * mgr,
+                              const uint8_t         * torrentHash,
+                              const struct in_addr  * addr );
 
 void tr_peerMgrAddIncoming( tr_peerMgr     * manager,

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 */
@@ -55 +54 @@
     BT_CANCEL               = 8,
     BT_PORT                 = 9,
-    BT_SUGGEST              = 13,
-    BT_HAVE_ALL             = 14,
-    BT_HAVE_NONE            = 15,
-    BT_REJECT               = 16,
-    BT_ALLOWED_FAST         = 17,
+
+    BT_FEXT_SUGGEST         = 13,
+    BT_FEXT_HAVE_ALL        = 14,
+    BT_FEXT_HAVE_NONE       = 15,
+    BT_FEXT_REJECT          = 16,
+    BT_FEXT_ALLOWED_FAST    = 17,
+
     BT_LTEP                 = 20,
 
@@ -65 +66 @@
 
     TR_LTEP_PEX             = 1,
+
+
 
     MIN_CHOKE_PERIOD_SEC    = ( 10 ),
@@ -72 +75 @@
 
     PEX_INTERVAL            = ( 90 * 1000 ), /* msec between sendPex() calls */
-    PEER_PULSE_INTERVAL     = ( 250 ),       /* msec between peerPulse() calls
-                                               */
 
     MAX_QUEUE_SIZE          = ( 100 ),
-
-    /* (fast peers) max number of pieces we fast-allow to another peer */
-    MAX_FAST_ALLOWED_COUNT   = 10,
-
-    /* (fast peers) max threshold for allowing fast-pieces requests */
-    MAX_FAST_ALLOWED_THRESHOLD = 10,
 
     /* how long an unsent request can stay queued before it's returned
@@ -98 +93 @@
     /* number of pieces to remove from the bitfield when
      * lazy bitfields are turned on */
-    LAZY_PIECE_COUNT = 26
+    LAZY_PIECE_COUNT = 26,
+
+    /* number of pieces we'll allow in our fast set */
+    MAX_FAST_SET_SIZE = 3
 };
 
@@ -122 +120 @@
 
 static int
-compareRequest( const void * va,
-                const void * vb )
+compareRequest( const void * va, const void * vb )
 {
     const struct peer_request * a = va;
@@ -170 +167 @@
 
 static void
-reqListCopy( struct request_list *       dest,
-             const struct request_list * src )
+reqListCopy( struct request_list * dest, const struct request_list * src )
 {
     dest->count = dest->max = src->count;
-    dest->requests =
-        tr_memdup( src->requests, dest->count * sizeof( struct peer_request ) );
+    dest->requests = tr_memdup( src->requests, dest->count * sizeof( struct peer_request ) );
 }
 
@@ -185 +180 @@
 
     memmove( &list->requests[i],
-            &list->requests[i + 1],
-            sizeof( struct peer_request ) * ( --list->count - i ) );
+             &list->requests[i + 1],
+             sizeof( struct peer_request ) * ( --list->count - i ) );
 }
 
@@ -260 +255 @@
 };
 
+/**
+ * Low-level communication state information about a connected peer.
+ *
+ * This structure remembers the low-level protocol states that we're
+ * in with this peer, such as active requests, pex messages, and so on.
+ * Its fields are all private to peer-msgs.c.
+ *
+ * Data not directly involved with sending & receiving messages is
+ * stored in tr_peer, where it can be accessed by both peermsgs and
+ * the peer manager.
+ *
+ * @see struct peer_atom
+ * @see tr_peer
+ */
 struct tr_peermsgs
 {
@@ -266 +275 @@
     tr_bool         clientSentLtepHandshake;
     tr_bool         peerSentLtepHandshake;
+    tr_bool         haveFastSet;
 
     uint8_t         state;
     uint8_t         ut_pex_id;
     uint16_t        pexCount;
-    uint16_t        minActiveRequests;
     uint16_t        maxActiveRequests;
+
+    size_t                 fastsetSize;
+    tr_piece_index_t       fastset[MAX_FAST_SET_SIZE];
 
     /* how long the outMessages batch should be allowed to grow before
@@ -278 +290 @@
     int                    outMessagesBatchPeriod;
 
-    tr_peer *              info;
+    tr_peer *              peer;
 
     tr_session *           session;
     tr_torrent *           torrent;
-    tr_peerIo *            io;
 
     tr_publisher_t *       publisher;
@@ -289 +300 @@
 
     struct request_list    peerAskedFor;
-    struct request_list    peerAskedForFast;
     struct request_list    clientAskedFor;
     struct request_list    clientWillAskFor;
 
-    tr_timer *             pexTimer;
+    tr_timer             * pexTimer;
+    tr_pex               * pex;
 
     time_t                 clientSentPexAt;
@@ -301 +312 @@
     time_t                outMessagesBatchedAt;
 
-    tr_bitfield *         peerAllowedPieces;
-
     struct tr_incoming    incoming;
-
-    tr_pex *              pex;
 
     /* if the peer supports the Extension Protocol in BEP 10 and
@@ -318 +325 @@
 
 static void
-myDebug( const char *               file,
-         int                        line,
+myDebug( const char * file, int line,
          const struct tr_peermsgs * msgs,
-         const char *               fmt,
-         ... )
+         const char * fmt, ... )
 {
     FILE * fp = tr_getLog( );
@@ -336 +341 @@
                              tr_getLogTimeStr( timestr, sizeof( timestr ) ),
                              msgs->torrent->info.name,
-                             tr_peerIoGetAddrStr( msgs->io ),
-                             msgs->info->client );
+                             tr_peerIoGetAddrStr( msgs->peer->io ),
+                             msgs->peer->client );
         va_start( args, fmt );
         evbuffer_add_vprintf( buf, fmt, args );
@@ -371 +376 @@
 
 static void
-protocolSendRequest( tr_peermsgs *               msgs,
+dbgOutMessageLen( tr_peermsgs * msgs )
+{
+    dbgmsg( msgs, "outMessage size is now %zu", EVBUFFER_LENGTH( msgs->outMessages ) );
+}
+
+static void
+protocolSendReject( tr_peermsgs * msgs, const struct peer_request * req )
+{
+    tr_peerIo       * io  = msgs->peer->io;
+    struct evbuffer * out = msgs->outMessages;
+
+    assert( tr_peerIoSupportsFEXT( msgs->peer->io ) );
+
+    tr_peerIoWriteUint32( io, out, sizeof( uint8_t ) + 3 * sizeof( uint32_t ) );
+    tr_peerIoWriteUint8 ( io, out, BT_FEXT_REJECT );
+    tr_peerIoWriteUint32( io, out, req->index );
+    tr_peerIoWriteUint32( io, out, req->offset );
+    tr_peerIoWriteUint32( io, out, req->length );
+
+    dbgmsg( msgs, "rejecting %u:%u->%u...", req->index, req->offset, req->length );
+    dbgOutMessageLen( msgs );
+}
+
+static void
+protocolSendRequest( tr_peermsgs               * msgs,
                      const struct peer_request * req )
 {
-    tr_peerIo *       io = msgs->io;
+    tr_peerIo       * io  = msgs->peer->io;
     struct evbuffer * out = msgs->outMessages;
 
@@ -382 +411 @@
     tr_peerIoWriteUint32( io, out, req->offset );
     tr_peerIoWriteUint32( io, out, req->length );
-    dbgmsg( msgs, "requesting %u:%u->%u... outMessage size is now %d",
-           req->index, req->offset, req->length, (int)EVBUFFER_LENGTH( out ) );
+
+    dbgmsg( msgs, "requesting %u:%u->%u...", req->index, req->offset, req->length );
+    dbgOutMessageLen( msgs );
     pokeBatchPeriod( msgs, IMMEDIATE_PRIORITY_INTERVAL_SECS );
 }
 
 static void
-protocolSendCancel( tr_peermsgs *               msgs,
+protocolSendCancel( tr_peermsgs               * msgs,
                     const struct peer_request * req )
 {
-    tr_peerIo *       io = msgs->io;
+    tr_peerIo       * io  = msgs->peer->io;
     struct evbuffer * out = msgs->outMessages;
 
@@ -399 +429 @@
     tr_peerIoWriteUint32( io, out, req->offset );
     tr_peerIoWriteUint32( io, out, req->length );
-    dbgmsg( msgs, "cancelling %u:%u->%u... outMessage size is now %d",
-           req->index, req->offset, req->length, (int)EVBUFFER_LENGTH( out ) );
+
+    dbgmsg( msgs, "cancelling %u:%u->%u...", req->index, req->offset, req->length );
+    dbgOutMessageLen( msgs );
     pokeBatchPeriod( msgs, IMMEDIATE_PRIORITY_INTERVAL_SECS );
 }
@@ -408 +439 @@
                   uint32_t      index )
 {
-    tr_peerIo *       io = msgs->io;
+    tr_peerIo       * io  = msgs->peer->io;
     struct evbuffer * out = msgs->outMessages;
 
-    tr_peerIoWriteUint32( io, out, sizeof( uint8_t ) + sizeof( uint32_t ) );
+    tr_peerIoWriteUint32( io, out, sizeof(uint8_t) + sizeof(uint32_t) );
     tr_peerIoWriteUint8 ( io, out, BT_HAVE );
     tr_peerIoWriteUint32( io, out, index );
-    dbgmsg( msgs, "sending Have %u.. outMessage size is now %d",
-           index, (int)EVBUFFER_LENGTH( out ) );
+
+    dbgmsg( msgs, "sending Have %u", index );
+    dbgOutMessageLen( msgs );
     pokeBatchPeriod( msgs, LOW_PRIORITY_INTERVAL_SECS );
 }
+
+#if 0
+static void
+protocolSendAllowedFast( tr_peermsgs * msgs, uint32_t pieceIndex )
+{
+    tr_peerIo       * io  = msgs->peer->io;
+    struct evbuffer * out = msgs->outMessages;
+
+    assert( tr_peerIoSupportsFEXT( msgs->peer->io ) );
+
+    tr_peerIoWriteUint32( io, out, sizeof(uint8_t) + sizeof(uint32_t) );
+    tr_peerIoWriteUint8 ( io, out, BT_FEXT_ALLOWED_FAST );
+    tr_peerIoWriteUint32( io, out, pieceIndex );
+
+    dbgmsg( msgs, "sending Allowed Fast %u...", pieceIndex );
+    dbgOutMessageLen( msgs );
+}
+#endif
 
 static void
@@ -423 +473 @@
                    int           choke )
 {
-    tr_peerIo *       io = msgs->io;
+    tr_peerIo       * io  = msgs->peer->io;
     struct evbuffer * out = msgs->outMessages;
 
     tr_peerIoWriteUint32( io, out, sizeof( uint8_t ) );
     tr_peerIoWriteUint8 ( io, out, choke ? BT_CHOKE : BT_UNCHOKE );
-    dbgmsg( msgs, "sending %s... outMessage size is now %d",
-           ( choke ? "Choke" : "Unchoke" ),
-           (int)EVBUFFER_LENGTH( out ) );
+
+    dbgmsg( msgs, "sending %s...", choke ? "Choke" : "Unchoke" );
+    dbgOutMessageLen( msgs );
+    pokeBatchPeriod( msgs, IMMEDIATE_PRIORITY_INTERVAL_SECS );
+}
+
+static void
+protocolSendHaveAll( tr_peermsgs * msgs )
+{
+    tr_peerIo       * io  = msgs->peer->io;
+    struct evbuffer * out = msgs->outMessages;
+
+    assert( tr_peerIoSupportsFEXT( msgs->peer->io ) );
+
+    tr_peerIoWriteUint32( io, out, sizeof( uint8_t ) );
+    tr_peerIoWriteUint8 ( io, out, BT_FEXT_HAVE_ALL );
+
+    dbgmsg( msgs, "sending HAVE_ALL..." );
+    dbgOutMessageLen( msgs );
+    pokeBatchPeriod( msgs, IMMEDIATE_PRIORITY_INTERVAL_SECS );
+}
+
+static void
+protocolSendHaveNone( tr_peermsgs * msgs )
+{
+    tr_peerIo       * io  = msgs->peer->io;
+    struct evbuffer * out = msgs->outMessages;
+
+    assert( tr_peerIoSupportsFEXT( msgs->peer->io ) );
+
+    tr_peerIoWriteUint32( io, out, sizeof( uint8_t ) );
+    tr_peerIoWriteUint8 ( io, out, BT_FEXT_HAVE_NONE );
+
+    dbgmsg( msgs, "sending HAVE_NONE..." );
+    dbgOutMessageLen( msgs );
     pokeBatchPeriod( msgs, IMMEDIATE_PRIORITY_INTERVAL_SECS );
 }
@@ -438 +520 @@
 **/
 
-static const tr_peer_event blankEvent = { 0, 0, 0, 0, 0.0f, 0, 0 };
-
-static void
-publish( tr_peermsgs *   msgs,
-         tr_peer_event * e )
-{
-    tr_publisherPublish( msgs->publisher, msgs->info, e );
-}
-
-static void
-fireError( tr_peermsgs * msgs,
-           int           err )
+static const tr_peer_event blankEvent = { 0, 0, 0, 0, 0.0f, 0, 0, 0 };
+
+static void
+publish( tr_peermsgs * msgs, tr_peer_event * e )
+{
+    assert( msgs->peer );
+    assert( msgs->peer->msgs == msgs );
+
+    tr_publisherPublish( msgs->publisher, msgs->peer, e );
+}
+
+static void
+fireError( tr_peermsgs * msgs, int err )
 {
     tr_peer_event e = blankEvent;
-
     e.eventType = TR_PEER_ERROR;
     e.err = err;
@@ -459 +541 @@
 
 static void
+fireUploadOnly( tr_peermsgs * msgs, tr_bool uploadOnly )
+{
+    tr_peer_event e = blankEvent;
+    e.eventType = TR_PEER_UPLOAD_ONLY;
+    e.uploadOnly = uploadOnly;
+    publish( msgs, &e );
+}
+
+static void
 fireNeedReq( tr_peermsgs * msgs )
 {
     tr_peer_event e = blankEvent;
-
     e.eventType = TR_PEER_NEED_REQ;
     publish( msgs, &e );
@@ -471 +561 @@
 {
     tr_peer_event e = blankEvent;
-
     e.eventType = TR_PEER_PEER_PROGRESS;
-    e.progress = msgs->info->progress;
+    e.progress = msgs->peer->progress;
     publish( msgs, &e );
 }
 
 static void
-fireGotBlock( tr_peermsgs *               msgs,
-              const struct peer_request * req )
+fireGotBlock( tr_peermsgs * msgs, const struct peer_request * req )
 {
     tr_peer_event e = blankEvent;
-
     e.eventType = TR_PEER_CLIENT_GOT_BLOCK;
     e.pieceIndex = req->index;
@@ -500 +587 @@
     e.eventType = TR_PEER_CLIENT_GOT_DATA;
     e.wasPieceData = wasPieceData;
+    publish( msgs, &e );
+}
+
+static void
+fireClientGotSuggest( tr_peermsgs * msgs, uint32_t pieceIndex )
+{
+    tr_peer_event e = blankEvent;
+    e.eventType = TR_PEER_CLIENT_GOT_SUGGEST;
+    e.pieceIndex = pieceIndex;
+    publish( msgs, &e );
+}
+
+static void
+fireClientGotAllowedFast( tr_peermsgs * msgs, uint32_t pieceIndex )
+{
+    tr_peer_event e = blankEvent;
+    e.eventType = TR_PEER_CLIENT_GOT_ALLOWED_FAST;
+    e.pieceIndex = pieceIndex;
     publish( msgs, &e );
 }
@@ -529 +634 @@
 
 /**
+***  ALLOWED FAST SET
+***  For explanation, see http://www.bittorrent.org/beps/bep_0006.html
+**/
+
+size_t
+tr_generateAllowedSet( tr_piece_index_t * setmePieces,
+                       size_t             desiredSetSize,
+                       size_t             pieceCount,
+                       const uint8_t    * infohash,
+                       const tr_address * addr )
+{
+    size_t setSize = 0;
+
+    assert( setmePieces );
+    assert( desiredSetSize <= pieceCount );
+    assert( desiredSetSize );
+    assert( pieceCount );
+    assert( infohash );
+    assert( addr );
+
+    if( 1 )
+    {
+        uint8_t w[SHA_DIGEST_LENGTH + 4];
+        uint8_t x[SHA_DIGEST_LENGTH];
+
+        *(uint32_t*)w = ntohl( htonl( addr->s_addr ) & 0xffffff00 ); /* (1) */
+        memcpy( w + 4, infohash, SHA_DIGEST_LENGTH );                /* (2) */
+        tr_sha1( x, w, sizeof( w ), NULL );                          /* (3) */
+
+        while( setSize<desiredSetSize )
+        {
+            int i;
+            for( i=0; i<5 && setSize<desiredSetSize; ++i )           /* (4) */
+            {
+                size_t k;
+                uint32_t j = i * 4;                                  /* (5) */
+                uint32_t y = ntohl( *( uint32_t* )( x + j ) );       /* (6) */
+                uint32_t index = y % pieceCount;                     /* (7) */
+
+                for( k=0; k<setSize; ++k )                           /* (8) */
+                    if( setmePieces[k] == index )
+                        break;
+
+                if( k == setSize )
+                    setmePieces[setSize++] = index;                  /* (9) */
+            }
+
+            tr_sha1( x, x, sizeof( x ), NULL );                      /* (3) */
+        }
+    }
+
+    return setSize;
+}
+
+static void
+updateFastSet( tr_peermsgs * msgs UNUSED )
+{
+#if 0
+    const tr_bool fext = tr_peerIoSupportsFEXT( msgs->peer->io );
+    const int peerIsNeedy = msgs->peer->progress < 0.10;
+
+    if( fext && peerIsNeedy && !msgs->haveFastSet )
+    {
+        size_t i;
+        const struct tr_address * addr = tr_peerIoGetAddress( msgs->peer->io, NULL );
+        const tr_info * inf = &msgs->torrent->info;
+        const size_t numwant = MIN( MAX_FAST_SET_SIZE, inf->pieceCount );
+
+        /* build the fast set */
+        msgs->fastsetSize = tr_generateAllowedSet( msgs->fastset, numwant, inf->pieceCount, inf->hash, addr );
+        msgs->haveFastSet = 1;
+
+        /* send it to the peer */
+        for( i=0; i<msgs->fastsetSize; ++i )
+            protocolSendAllowedFast( msgs, msgs->fastset[i] );
+    }
+#endif
+}
+
+/**
 ***  INTEREST
 **/
 
 static int
-isPieceInteresting( const tr_peermsgs * peer,
+isPieceInteresting( const tr_peermsgs * msgs,
                     tr_piece_index_t    piece )
 {
-    const tr_torrent * torrent = peer->torrent;
+    const tr_torrent * torrent = msgs->torrent;
 
     return ( !torrent->info.pieces[piece].dnd )                 /* we want it */
-           && ( !tr_cpPieceIsComplete( torrent->completion, piece ) ) /* !have
-                                                                        */
-           && ( tr_bitfieldHas( peer->info->have, piece ) );   /* peer has it */
+          && ( !tr_cpPieceIsComplete( torrent->completion, piece ) ) /* !have */
+          && ( tr_bitfieldHas( msgs->peer->have, piece ) );    /* peer has it */
 }
 
@@ -562 +746 @@
     bitfield = tr_cpPieceBitfield( torrent->completion );
 
-    if( !msgs->info->have )
+    if( !msgs->peer->have )
         return TRUE;
 
-    assert( bitfield->byteCount == msgs->info->have->byteCount );
+    assert( bitfield->byteCount == msgs->peer->have->byteCount );
 
     for( i = 0; i < torrent->info.pieceCount; ++i )
@@ -583 +767 @@
     assert( weAreInterested == 0 || weAreInterested == 1 );
 
-    msgs->info->clientIsInterested = weAreInterested;
-    dbgmsg( msgs, "Sending %s",
-            weAreInterested ? "Interested" : "Not Interested" );
-    tr_peerIoWriteUint32( msgs->io, out, sizeof( uint8_t ) );
-    tr_peerIoWriteUint8 (
-        msgs->io, out, weAreInterested ? BT_INTERESTED : BT_NOT_INTERESTED );
+    msgs->peer->clientIsInterested = weAreInterested;
+    dbgmsg( msgs, "Sending %s", weAreInterested ? "Interested" : "Not Interested" );
+    tr_peerIoWriteUint32( msgs->peer->io, out, sizeof( uint8_t ) );
+    tr_peerIoWriteUint8 ( msgs->peer->io, out, weAreInterested ? BT_INTERESTED : BT_NOT_INTERESTED );
+
     pokeBatchPeriod( msgs, HIGH_PRIORITY_INTERVAL_SECS );
-    dbgmsg( msgs, "outMessage size is now %d", (int)EVBUFFER_LENGTH( out ) );
+    dbgOutMessageLen( msgs );
 }
 
@@ -598 +781 @@
     const int i = isPeerInteresting( msgs );
 
-    if( i != msgs->info->clientIsInterested )
+    if( i != msgs->peer->clientIsInterested )
         sendInterest( msgs, i );
     if( i )
@@ -604 +787 @@
 }
 
-static void
-cancelAllRequestsToClientExceptFast( tr_peermsgs * msgs )
-{
-    reqListClear( &msgs->peerAskedFor );
+static int
+popNextRequest( tr_peermsgs *         msgs,
+                struct peer_request * setme )
+{
+    return reqListPop( &msgs->peerAskedFor, setme );
+}
+
+static void
+cancelAllRequestsToClient( tr_peermsgs * msgs )
+{
+    struct peer_request req;
+    const int mustSendCancel = tr_peerIoSupportsFEXT( msgs->peer->io );
+
+    while( popNextRequest( msgs, &req ) )
+        if( mustSendCancel )
+            protocolSendReject( msgs, &req );
 }
 
@@ -618 +813 @@
 
     assert( msgs );
-    assert( msgs->info );
+    assert( msgs->peer );
     assert( choke == 0 || choke == 1 );
 
-    if( msgs->info->chokeChangedAt > fibrillationTime )
-    {
-        dbgmsg( msgs, "Not changing choke to %d to avoid fibrillation",
-                choke );
-    }
-    else if( msgs->info->peerIsChoked != choke )
-    {
-        msgs->info->peerIsChoked = choke;
+    if( msgs->peer->chokeChangedAt > fibrillationTime )
+    {
+        dbgmsg( msgs, "Not changing choke to %d to avoid fibrillation", choke );
+    }
+    else if( msgs->peer->peerIsChoked != choke )
+    {
+        msgs->peer->peerIsChoked = choke;
         if( choke )
-            cancelAllRequestsToClientExceptFast( msgs );
+            cancelAllRequestsToClient( msgs );
         protocolSendChoke( msgs, choke );
-        msgs->info->chokeChangedAt = now;
+        msgs->peer->chokeChangedAt = now;
     }
 }
@@ -648 +842 @@
     /* since we have more pieces now, we might not be interested in this peer */
     updateInterest( msgs );
-}
-
-#if 0
-static void
-sendFastSuggest( tr_peermsgs * msgs,
-                 uint32_t      pieceIndex )
-{
-    assert( msgs );
-
-    if( tr_peerIoSupportsFEXT( msgs->io ) )
-    {
-        tr_peerIoWriteUint32( msgs->io, msgs->outMessages,
-                             sizeof( uint8_t ) + sizeof( uint32_t ) );
-        tr_peerIoWriteUint8( msgs->io, msgs->outMessages, BT_SUGGEST );
-        tr_peerIoWriteUint32( msgs->io, msgs->outMessages, pieceIndex );
-    }
-}
-
-static void
-sendFastHave( tr_peermsgs * msgs,
-              int           all )
-{
-    assert( msgs );
-
-    if( tr_peerIoSupportsFEXT( msgs->io ) )
-    {
-        tr_peerIoWriteUint32( msgs->io, msgs->outMessages, sizeof( uint8_t ) );
-        tr_peerIoWriteUint8( msgs->io, msgs->outMessages,
-                            ( all ? BT_HAVE_ALL
-                              : BT_HAVE_NONE ) );
-        updateInterest( msgs );
-    }
-}
-
-#endif
-
-static void
-sendFastReject( tr_peermsgs * msgs,
-                uint32_t      pieceIndex,
-                uint32_t      offset,
-                uint32_t      length )
-{
-    assert( msgs );
-
-    if( tr_peerIoSupportsFEXT( msgs->io ) )
-    {
-        struct evbuffer * out = msgs->outMessages;
-        const uint32_t    len = sizeof( uint8_t ) + 3 * sizeof( uint32_t );
-        dbgmsg( msgs, "sending fast reject %u:%u->%u", pieceIndex, offset,
-                length );
-        tr_peerIoWriteUint32( msgs->io, out, len );
-        tr_peerIoWriteUint8( msgs->io, out, BT_REJECT );
-        tr_peerIoWriteUint32( msgs->io, out, pieceIndex );
-        tr_peerIoWriteUint32( msgs->io, out, offset );
-        tr_peerIoWriteUint32( msgs->io, out, length );
-        pokeBatchPeriod( msgs, LOW_PRIORITY_INTERVAL_SECS );
-        dbgmsg( msgs, "outMessage size is now %d",
-               (int)EVBUFFER_LENGTH( out ) );
-    }
-}
-
-static tr_bitfield*
-getPeerAllowedPieces( tr_peermsgs * msgs )
-{
-    if( !msgs->peerAllowedPieces && tr_peerIoSupportsFEXT( msgs->io ) )
-    {
-        msgs->peerAllowedPieces = tr_peerMgrGenerateAllowedSet(
-            MAX_FAST_ALLOWED_COUNT,
-            msgs->torrent->info.pieceCount,
-            msgs->torrent->info.hash,
-            tr_peerIoGetAddress( msgs->io, NULL ) );
-    }
-
-    return msgs->peerAllowedPieces;
-}
-
-static void
-sendFastAllowed( tr_peermsgs * msgs,
-                 uint32_t      pieceIndex )
-{
-    assert( msgs );
-
-    if( tr_peerIoSupportsFEXT( msgs->io ) )
-    {
-        struct evbuffer * out = msgs->outMessages;
-        dbgmsg( msgs, "sending fast allowed" );
-        tr_peerIoWriteUint32( msgs->io, out,  sizeof( uint8_t ) +
-                             sizeof( uint32_t ) );
-        tr_peerIoWriteUint8( msgs->io, out, BT_ALLOWED_FAST );
-        tr_peerIoWriteUint32( msgs->io, out, pieceIndex );
-        pokeBatchPeriod( msgs, LOW_PRIORITY_INTERVAL_SECS );
-        dbgmsg( msgs, "outMessage size is now %d",
-               (int)EVBUFFER_LENGTH( out ) );
-    }
-}
-
-static void
-sendFastAllowedSet( tr_peermsgs * msgs )
-{
-    tr_piece_index_t i = 0;
-
-    while( i <= msgs->torrent->info.pieceCount )
-    {
-        if( tr_bitfieldHas( getPeerAllowedPieces( msgs ), i ) )
-            sendFastAllowed( msgs, i );
-        i++;
-    }
-}
-
-static void
-maybeSendFastAllowedSet( tr_peermsgs * msgs )
-{
-    if( tr_bitfieldCountTrueBits( msgs->info->have ) <=
-        MAX_FAST_ALLOWED_THRESHOLD )
-        sendFastAllowedSet( msgs );
 }
 
@@ -787 +866 @@
 expireOldRequests( tr_peermsgs * msgs, const time_t now  )
 {
-    int                 i;
-    time_t              oldestAllowed;
+    int i;
+    time_t oldestAllowed;
     struct request_list tmp = REQUEST_LIST_INIT;
+    const tr_bool fext = tr_peerIoSupportsFEXT( msgs->peer->io );
+    dbgmsg( msgs, "entering `expire old requests' block" );
 
     /* cancel requests that have been queued for too long */
     oldestAllowed = now - QUEUED_REQUEST_TTL_SECS;
     reqListCopy( &tmp, &msgs->clientWillAskFor );
-    for( i = 0; i < tmp.count; ++i )
-    {
+    for( i=0; i<tmp.count; ++i ) {
         const struct peer_request * req = &tmp.requests[i];
         if( req->time_requested < oldestAllowed )
@@ -802 +882 @@
     reqListClear( &tmp );
 
-    /* cancel requests that were sent too long ago */
-    oldestAllowed = now - SENT_REQUEST_TTL_SECS;
-    reqListCopy( &tmp, &msgs->clientAskedFor );
-    for( i = 0; i < tmp.count; ++i )
-    {
-        const struct peer_request * req = &tmp.requests[i];
-        if( req->time_requested < oldestAllowed )
-            tr_peerMsgsCancel( msgs, req->index, req->offset, req->length );
-    }
-    reqListClear( &tmp );
+    /* if the peer doesn't support "Reject Request",
+     * cancel requests that were sent too long ago. */
+    if( !fext ) {
+        oldestAllowed = now - SENT_REQUEST_TTL_SECS;
+        reqListCopy( &tmp, &msgs->clientAskedFor );
+        for( i=0; i<tmp.count; ++i ) {
+            const struct peer_request * req = &tmp.requests[i];
+            if( req->time_requested < oldestAllowed )
+                tr_peerMsgsCancel( msgs, req->index, req->offset, req->length );
+        }
+        reqListClear( &tmp );
+    }
+
+    dbgmsg( msgs, "leaving `expire old requests' block" );
 }
 
@@ -818 +902 @@
 {
     const int           max = msgs->maxActiveRequests;
-    const int           min = msgs->minActiveRequests;
     int                 sent = 0;
     int                 count = msgs->clientAskedFor.count;
     struct peer_request req;
 
-    if( count > min )
-        return;
-    if( msgs->info->clientIsChoked )
+    if( msgs->peer->clientIsChoked )
         return;
     if( !tr_torrentIsPieceTransferAllowed( msgs->torrent, TR_PEER_TO_CLIENT ) )
@@ -832 +913 @@
     while( ( count < max ) && reqListPop( &msgs->clientWillAskFor, &req ) )
     {
-        const tr_block_index_t block =
-            _tr_block( msgs->torrent, req.index, req.offset );
+        const tr_block_index_t block = _tr_block( msgs->torrent, req.index, req.offset );
 
         assert( requestIsValid( msgs, &req ) );
-        assert( tr_bitfieldHas( msgs->info->have, req.index ) );
+        assert( tr_bitfieldHas( msgs->peer->have, req.index ) );
 
         /* don't ask for it if we've already got it... this block may have
@@ -852 +932 @@
 
     if( sent )
-        dbgmsg( msgs,
-                "pump sent %d requests, now have %d active and %d queued",
-                sent,
-                msgs->clientAskedFor.count,
-                msgs->clientWillAskFor.count );
+        dbgmsg( msgs, "pump sent %d requests, now have %d active and %d queued",
+                sent, msgs->clientAskedFor.count, msgs->clientWillAskFor.count );
 
     if( count < max )
@@ -886 +963 @@
 
     /* don't send requests to choked clients */
-    if( msgs->info->clientIsChoked )
-    {
+    if( msgs->peer->clientIsChoked ) {
         dbgmsg( msgs, "declining request because they're choking us" );
         return TR_ADDREQ_CLIENT_CHOKED;
@@ -893 +969 @@
 
     /* peer doesn't have this piece */
-    if( !tr_bitfieldHas( msgs->info->have, index ) )
+    if( !tr_bitfieldHas( msgs->peer->have, index ) )
         return TR_ADDREQ_MISSING;
 
@@ -929 +1005 @@
 cancelAllRequestsToPeer( tr_peermsgs * msgs, tr_bool sendCancel )
 {
-    int                 i;
+    int i;
     struct request_list a = msgs->clientWillAskFor;
     struct request_list b = msgs->clientAskedFor;
@@ -969 +1045 @@
     /* if it's only in the queue and hasn't been sent yet, free it */
     if( reqListRemove( &msgs->clientWillAskFor, &req ) ) {
-        dbgmsg( msgs, "cancelling %"PRIu32":%"PRIu32"->%"PRIu32"\n", pieceIndex, offset, length );
+        dbgmsg( msgs, "cancelling %"PRIu32":%"PRIu32"->%"PRIu32, pieceIndex, offset, length );
         fireCancelledReq( msgs, &req );
     }
@@ -975 +1051 @@
     /* 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 );
+        dbgmsg( msgs, "cancelling %"PRIu32":%"PRIu32"->%"PRIu32, pieceIndex, offset, length );
         protocolSendCancel( msgs, &req );
         fireCancelledReq( msgs, &req );
     }
 }
+
 
 /**
@@ -988 +1065 @@
 sendLtepHandshake( tr_peermsgs * msgs )
 {
-    tr_benc           val, *m;
-    char *            buf;
-    int               len;
-    int               pex;
+    tr_benc val, *m;
+    char * buf;
+    int len;
+    int pex;
     struct evbuffer * out = msgs->outMessages;
 
@@ -1008 +1085 @@
         pex = 1;
 
-    tr_bencInitDict( &val, 4 );
-    tr_bencDictAddInt( &val, "e",
-                       msgs->session->encryptionMode != TR_CLEAR_PREFERRED );
+    tr_bencInitDict( &val, 5 );
+    tr_bencDictAddInt( &val, "e", msgs->session->encryptionMode != TR_CLEAR_PREFERRED );
     tr_bencDictAddInt( &val, "p", tr_sessionGetPeerPort( msgs->session ) );
+    tr_bencDictAddInt( &val, "upload_only", tr_torrentIsSeed( msgs->torrent ) );
     tr_bencDictAddStr( &val, "v", TR_NAME " " USERAGENT_PREFIX );
     m  = tr_bencDictAddDict( &val, "m", 1 );
@@ -1018 +1095 @@
     buf = tr_bencSave( &val, &len );
 
-    tr_peerIoWriteUint32( msgs->io, out, 2 * sizeof( uint8_t ) + len );
-    tr_peerIoWriteUint8 ( msgs->io, out, BT_LTEP );
-    tr_peerIoWriteUint8 ( msgs->io, out, LTEP_HANDSHAKE );
-    tr_peerIoWriteBytes ( msgs->io, out, buf, len );
+    tr_peerIoWriteUint32( msgs->peer->io, out, 2 * sizeof( uint8_t ) + len );
+    tr_peerIoWriteUint8 ( msgs->peer->io, out, BT_LTEP );
+    tr_peerIoWriteUint8 ( msgs->peer->io, out, LTEP_HANDSHAKE );
+    tr_peerIoWriteBytes ( msgs->peer->io, out, buf, len );
     pokeBatchPeriod( msgs, IMMEDIATE_PRIORITY_INTERVAL_SECS );
-    dbgmsg( msgs, "outMessage size is now %d", (int)EVBUFFER_LENGTH( out ) );
+    dbgOutMessageLen( msgs );
 
     /* cleanup */
@@ -1039 +1116 @@
     uint8_t * tmp = tr_new( uint8_t, len );
 
-    tr_peerIoReadBytes( msgs->io, inbuf, tmp, len );
+    tr_peerIoReadBytes( msgs->peer->io, inbuf, tmp, len );
     msgs->peerSentLtepHandshake = 1;
 
-    if( tr_bencLoad( tmp, len, &val, NULL ) || val.type != TYPE_DICT )
+    if( tr_bencLoad( tmp, len, &val, NULL ) || !tr_bencIsDict( &val ) )
     {
         dbgmsg( msgs, "GET  extended-handshake, couldn't get dictionary" );
@@ -1053 +1130 @@
     /* does the peer prefer encrypted connections? */
     if( tr_bencDictFindInt( &val, "e", &i ) )
-        msgs->info->encryption_preference = i ? ENCRYPTION_PREFERENCE_YES
-                                            : ENCRYPTION_PREFERENCE_NO;
+        msgs->peer->encryption_preference = i ? ENCRYPTION_PREFERENCE_YES
+                                              : ENCRYPTION_PREFERENCE_NO;
 
     /* check supported messages for utorrent pex */
     msgs->peerSupportsPex = 0;
-    if( tr_bencDictFindDict( &val, "m", &sub ) )
-    {
-        if( tr_bencDictFindInt( sub, "ut_pex", &i ) )
-        {
+    if( tr_bencDictFindDict( &val, "m", &sub ) ) {
+        if( tr_bencDictFindInt( sub, "ut_pex", &i ) ) {
             msgs->ut_pex_id = (uint8_t) i;
             msgs->peerSupportsPex = msgs->ut_pex_id == 0 ? 0 : 1;
@@ -1068 +1143 @@
     }
 
+    /* look for upload_only (BEP 21) */
+    if( tr_bencDictFindInt( &val, "upload_only", &i ) )
+        fireUploadOnly( msgs, i!=0 );
+
     /* get peer's listening port */
-    if( tr_bencDictFindInt( &val, "p", &i ) )
-    {
-        msgs->info->port = htons( (uint16_t)i );
-        dbgmsg( msgs, "msgs->port is now %hu", msgs->info->port );
+    if( tr_bencDictFindInt( &val, "p", &i ) ) {
+        msgs->peer->port = htons( (uint16_t)i );
+        dbgmsg( msgs, "msgs->port is now %hu", msgs->peer->port );
     }
 
@@ -1084 +1162 @@
 
 static void
-parseUtPex( tr_peermsgs *     msgs,
-            int               msglen,
-            struct evbuffer * inbuf )
-{
-    int                loaded = 0;
-    uint8_t *          tmp = tr_new( uint8_t, msglen );
-    tr_benc            val;
+parseUtPex( tr_peermsgs * msgs, int msglen, struct evbuffer * inbuf )
+{
+    int loaded = 0;
+    uint8_t * tmp = tr_new( uint8_t, msglen );
+    tr_benc val;
     const tr_torrent * tor = msgs->torrent;
-    const uint8_t *    added;
-    size_t             added_len;
-
-    tr_peerIoReadBytes( msgs->io, inbuf, tmp, msglen );
+    const uint8_t * added;
+    size_t added_len;
+
+    tr_peerIoReadBytes( msgs->peer->io, inbuf, tmp, msglen );
 
     if( tr_torrentAllowsPex( tor )
@@ -1129 +1205 @@
     uint8_t ltep_msgid;
 
-    tr_peerIoReadUint8( msgs->io, inbuf, &ltep_msgid );
+    tr_peerIoReadUint8( msgs->peer->io, inbuf, &ltep_msgid );
     msglen--;
 
@@ -1136 +1212 @@
         dbgmsg( msgs, "got ltep handshake" );
         parseLtepHandshake( msgs, msglen, inbuf );
-        if( tr_peerIoSupportsLTEP( msgs->io ) )
+        if( tr_peerIoSupportsLTEP( msgs->peer->io ) )
         {
             sendLtepHandshake( msgs );
@@ -1165 +1241 @@
         return READ_LATER;
 
-    tr_peerIoReadUint32( msgs->io, inbuf, &len );
+    tr_peerIoReadUint32( msgs->peer->io, inbuf, &len );
 
     if( len == 0 ) /* peer sent us a keepalive message */
@@ -1192 +1268 @@
         return READ_LATER;
 
-    tr_peerIoReadUint8( msgs->io, inbuf, &id );
+    tr_peerIoReadUint8( msgs->peer->io, inbuf, &id );
     msgs->incoming.id = id;
 
@@ -1211 +1287 @@
 updatePeerProgress( tr_peermsgs * msgs )
 {
-    msgs->info->progress = tr_bitfieldCountTrueBits( msgs->info->have )
-                           / (float)msgs->torrent->info.pieceCount;
-    dbgmsg( msgs, "peer progress is %f", msgs->info->progress );
+    msgs->peer->progress = tr_bitfieldCountTrueBits( msgs->peer->have ) / (float)msgs->torrent->info.pieceCount;
+    dbgmsg( msgs, "peer progress is %f", msgs->peer->progress );
+    updateFastSet( msgs );
     updateInterest( msgs );
     firePeerProgress( msgs );
 }
 
-static int
-clientCanSendFastBlock( const tr_peermsgs * msgs UNUSED )
-{
-    /* don't send a fast piece if peer has MAX_FAST_ALLOWED_THRESHOLD pieces */
-    if( tr_bitfieldCountTrueBits( msgs->info->have ) >
-        MAX_FAST_ALLOWED_THRESHOLD )
-        return FALSE;
-
-    /* ...or if we don't have ourself enough pieces */
-    if( tr_bitfieldCountTrueBits( tr_cpPieceBitfield( msgs->torrent->
-                                                      completion ) ) <
-        MAX_FAST_ALLOWED_THRESHOLD )
-        return FALSE;
-
-    /* Maybe a bandwidth limit ? */
-    return TRUE;
-}
-
 static void
 peerMadeRequest( tr_peermsgs *               msgs,
                  const struct peer_request * req )
 {
+    const tr_bool fext = tr_peerIoSupportsFEXT( msgs->peer->io );
     const int reqIsValid = requestIsValid( msgs, req );
-    const int clientHasPiece = reqIsValid && tr_cpPieceIsComplete(
-        msgs->torrent->completion, req->index );
-    const int peerIsChoked = msgs->info->peerIsChoked;
-    const int peerIsFast = tr_peerIoSupportsFEXT( msgs->io );
-    const int pieceIsFast = reqIsValid && tr_bitfieldHas(
-        getPeerAllowedPieces( msgs ), req->index );
-    const int canSendFast = clientCanSendFastBlock( msgs );
-
-    if( !reqIsValid ) /* bad request */
-    {
+    const int clientHasPiece = reqIsValid && tr_cpPieceIsComplete( msgs->torrent->completion, req->index );
+    const int peerIsChoked = msgs->peer->peerIsChoked;
+
+    int allow = FALSE;
+
+    if( !reqIsValid )
         dbgmsg( msgs, "rejecting an invalid request." );
-        sendFastReject( msgs, req->index, req->offset, req->length );
-    }
-    else if( !clientHasPiece ) /* we don't have it */
-    {
+    else if( !clientHasPiece )
         dbgmsg( msgs, "rejecting request for a piece we don't have." );
-        sendFastReject( msgs, req->index, req->offset, req->length );
-    }
-    else if( peerIsChoked && !peerIsFast ) /* doesn't he know he's choked? */
-    {
-        tr_peerMsgsSetChoke( msgs, 1 );
-        sendFastReject( msgs, req->index, req->offset, req->length );
-    }
-    else if( peerIsChoked && peerIsFast && ( !pieceIsFast || !canSendFast ) )
-    {
-        sendFastReject( msgs, req->index, req->offset, req->length );
-    }
-    else /* YAY */
-    {
-        if( peerIsFast && pieceIsFast )
-            reqListAppend( &msgs->peerAskedForFast, req );
-        else
-            reqListAppend( &msgs->peerAskedFor, req );
-    }
+    else if( peerIsChoked )
+        dbgmsg( msgs, "rejecting request from choked peer" );
+    else
+        allow = TRUE;
+
+    if( allow )
+        reqListAppend( &msgs->peerAskedFor, req );
+    else if( fext )
+        protocolSendReject( msgs, req );
 }
 
 static int
-messageLengthIsCorrect( const tr_peermsgs * msg,
-                        uint8_t             id,
-                        uint32_t            len )
+messageLengthIsCorrect( const tr_peermsgs * msg, uint8_t id, uint32_t len )
 {
     switch( id )
@@ -1288 +1329 @@
         case BT_INTERESTED:
         case BT_NOT_INTERESTED:
-        case BT_HAVE_ALL:
-        case BT_HAVE_NONE:
+        case BT_FEXT_HAVE_ALL:
+        case BT_FEXT_HAVE_NONE:
             return len == 1;
 
         case BT_HAVE:
-        case BT_SUGGEST:
-        case BT_ALLOWED_FAST:
+        case BT_FEXT_SUGGEST:
+        case BT_FEXT_ALLOWED_FAST:
             return len == 5;
 
@@ -1302 +1343 @@
         case BT_REQUEST:
         case BT_CANCEL:
-        case BT_REJECT:
+        case BT_FEXT_REJECT:
             return len == 13;
 
@@ -1339 +1380 @@
             return READ_LATER;
 
-        tr_peerIoReadUint32( msgs->io, inbuf, &req->index );
-        tr_peerIoReadUint32( msgs->io, inbuf, &req->offset );
+        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 );
+        dbgmsg( msgs, "got incoming block header %u:%u->%u", req->index, req->offset, req->length );
         return READ_NOW;
     }
     else
     {
-        int          err;
+        int err;
 
         /* read in another chunk of data */
-        const size_t nLeft = req->length - EVBUFFER_LENGTH(
-            msgs->incoming.block );
-        size_t       n = MIN( nLeft, inlen );
-        uint8_t *    buf = tr_new( uint8_t, n );
+        const size_t nLeft = req->length - EVBUFFER_LENGTH( msgs->incoming.block );
+        size_t n = MIN( nLeft, inlen );
+        uint8_t * buf = tr_new( uint8_t, n );
         assert( EVBUFFER_LENGTH( inbuf ) >= n );
-        tr_peerIoReadBytes( msgs->io, inbuf, buf, n );
+        tr_peerIoReadBytes( msgs->peer->io, inbuf, buf, n );
         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",
-               (int)n, req->index, req->offset, req->length,
+        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 )
@@ -1369 +1407 @@
 
         /* 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 ) );
+        evbuffer_drain( msgs->incoming.block, EVBUFFER_LENGTH( msgs->incoming.block ) );
         req->length = 0;
         msgs->state = AWAITING_BT_LENGTH;
         if( !err )
             return READ_NOW;
-        else
-        {
+        else {
             fireError( msgs, err );
             return READ_ERR;
@@ -1388 +1423 @@
 
 static int
-readBtMessage( tr_peermsgs *     msgs,
-               struct evbuffer * inbuf,
-               size_t            inlen )
+readBtMessage( tr_peermsgs * msgs, struct evbuffer * inbuf, size_t inlen )
 {
     uint32_t      ui32;
@@ -1396 +1429 @@
     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 */
@@ -1402 +1436 @@
         return READ_LATER;
 
-    dbgmsg( msgs, "got BT id %d, len %d, buffer size is %d", (int)id,
-            (int)msglen,
-            (int)inlen );
+    dbgmsg( msgs, "got BT id %d, len %d, buffer size is %zu", (int)id, (int)msglen, inlen );
 
     if( !messageLengthIsCorrect( msgs, id, msglen + 1 ) )
     {
-        dbgmsg( msgs, "bad packet - BT message #%d with a length of %d",
-                (int)id, (int)msglen );
+        dbgmsg( msgs, "bad packet - BT message #%d with a length of %d", (int)id, (int)msglen );
         fireError( msgs, EMSGSIZE );
         return READ_ERR;
@@ -1418 +1449 @@
         case BT_CHOKE:
             dbgmsg( msgs, "got Choke" );
-            msgs->info->clientIsChoked = 1;
-            cancelAllRequestsToPeer( msgs, FALSE );
+            msgs->peer->clientIsChoked = 1;
+            if( !fext )
+                cancelAllRequestsToPeer( msgs, FALSE );
             break;
 
         case BT_UNCHOKE:
             dbgmsg( msgs, "got Unchoke" );
-            msgs->info->clientIsChoked = 0;
+            msgs->peer->clientIsChoked = 0;
             fireNeedReq( msgs );
             break;
@@ -1430 +1462 @@
         case BT_INTERESTED:
             dbgmsg( msgs, "got Interested" );
-            msgs->info->peerIsInterested = 1;
+            msgs->peer->peerIsInterested = 1;
             break;
 
         case BT_NOT_INTERESTED:
             dbgmsg( msgs, "got Not Interested" );
-            msgs->info->peerIsInterested = 0;
+            msgs->peer->peerIsInterested = 0;
             break;
 
         case BT_HAVE:
-            tr_peerIoReadUint32( msgs->io, inbuf, &ui32 );
+            tr_peerIoReadUint32( msgs->peer->io, inbuf, &ui32 );
             dbgmsg( msgs, "got Have: %u", ui32 );
-            if( tr_bitfieldAdd( msgs->info->have, ui32 ) )
-            {
+            if( tr_bitfieldAdd( msgs->peer->have, ui32 ) ) {
                 fireError( msgs, ERANGE );
                 return READ_ERR;
@@ -1455 +1486 @@
             dbgmsg( msgs, "got a bitfield" );
             msgs->peerSentBitfield = 1;
-            tr_peerIoReadBytes( msgs->io, inbuf, msgs->info->have->bits,
-                                msglen );
+            tr_peerIoReadBytes( msgs->peer->io, inbuf, msgs->peer->have->bits, msglen );
             updatePeerProgress( msgs );
-            maybeSendFastAllowedSet( msgs );
             fireNeedReq( msgs );
             break;
@@ -1466 +1495 @@
         {
             struct peer_request r;
-            tr_peerIoReadUint32( msgs->io, inbuf, &r.index );
-            tr_peerIoReadUint32( msgs->io, inbuf, &r.offset );
-            tr_peerIoReadUint32( msgs->io, inbuf, &r.length );
-            dbgmsg( msgs, "got Request: %u:%u->%u", r.index, r.offset,
-                    r.length );
+            tr_peerIoReadUint32( msgs->peer->io, inbuf, &r.index );
+            tr_peerIoReadUint32( msgs->peer->io, inbuf, &r.offset );
+            tr_peerIoReadUint32( msgs->peer->io, inbuf, &r.length );
+            dbgmsg( msgs, "got Request: %u:%u->%u", r.index, r.offset, r.length );
             peerMadeRequest( msgs, &r );
             break;
@@ -1478 +1506 @@
         {
             struct peer_request r;
-            tr_peerIoReadUint32( msgs->io, inbuf, &r.index );
-            tr_peerIoReadUint32( msgs->io, inbuf, &r.offset );
-            tr_peerIoReadUint32( msgs->io, inbuf, &r.length );
-            dbgmsg( msgs, "got a Cancel %u:%u->%u", r.index, r.offset,
-                    r.length );
-            reqListRemove( &msgs->peerAskedForFast, &r );
-            reqListRemove( &msgs->peerAskedFor, &r );
+            tr_peerIoReadUint32( msgs->peer->io, inbuf, &r.index );
+            tr_peerIoReadUint32( msgs->peer->io, inbuf, &r.offset );
+            tr_peerIoReadUint32( msgs->peer->io, inbuf, &r.length );
+            dbgmsg( msgs, "got a Cancel %u:%u->%u", r.index, r.offset, r.length );
+            if( reqListRemove( &msgs->peerAskedFor, &r ) && fext )
+                protocolSendReject( msgs, &r );
             break;
         }
@@ -1494 +1521 @@
         case BT_PORT:
             dbgmsg( msgs, "Got a BT_PORT" );
-            tr_peerIoReadUint16( msgs->io, inbuf, &msgs->info->port );
+            tr_peerIoReadUint16( msgs->peer->io, inbuf, &msgs->peer->port );
             break;
 
-        case BT_SUGGEST:
-        {
-            dbgmsg( msgs, "Got a BT_SUGGEST" );
-            tr_peerIoReadUint32( msgs->io, inbuf, &ui32 );
-            /* we don't do anything with this yet */
+        case BT_FEXT_SUGGEST:
+            dbgmsg( msgs, "Got a BT_FEXT_SUGGEST" );
+            tr_peerIoReadUint32( msgs->peer->io, inbuf, &ui32 );
+            if( fext )
+                fireClientGotSuggest( msgs, ui32 );
+            else {
+                fireError( msgs, EMSGSIZE );
+                return READ_ERR;
+            }
             break;
-        }
-
-        case BT_HAVE_ALL:
-            dbgmsg( msgs, "Got a BT_HAVE_ALL" );
-            tr_bitfieldAddRange( msgs->info->have, 0,
-                                 msgs->torrent->info.pieceCount );
-            updatePeerProgress( msgs );
-            maybeSendFastAllowedSet( msgs );
+
+        case BT_FEXT_ALLOWED_FAST:
+            dbgmsg( msgs, "Got a BT_FEXT_ALLOWED_FAST" );
+            tr_peerIoReadUint32( msgs->peer->io, inbuf, &ui32 );
+            if( fext )
+                fireClientGotAllowedFast( msgs, ui32 );
+            else {
+                fireError( msgs, EMSGSIZE );
+                return READ_ERR;
+            }
             break;
 
-
-        case BT_HAVE_NONE:
-            dbgmsg( msgs, "Got a BT_HAVE_NONE" );
-            tr_bitfieldClear( msgs->info->have );
-            updatePeerProgress( msgs );
-            maybeSendFastAllowedSet( msgs );
+        case BT_FEXT_HAVE_ALL:
+            dbgmsg( msgs, "Got a BT_FEXT_HAVE_ALL" );
+            if( fext ) {
+                tr_bitfieldAddRange( msgs->peer->have, 0, msgs->torrent->info.pieceCount );
+                updatePeerProgress( msgs );
+            } else {
+                fireError( msgs, EMSGSIZE );
+                return READ_ERR;
+            }
             break;
 
-        case BT_REJECT:
+        case BT_FEXT_HAVE_NONE:
+            dbgmsg( msgs, "Got a BT_FEXT_HAVE_NONE" );
+            if( fext ) {
+                tr_bitfieldClear( msgs->peer->have );
+                updatePeerProgress( msgs );
+            } else {
+                fireError( msgs, EMSGSIZE );
+                return READ_ERR;
+            }
+            break;
+
+        case BT_FEXT_REJECT:
         {
             struct peer_request r;
-            dbgmsg( msgs, "Got a BT_REJECT" );
-            tr_peerIoReadUint32( msgs->io, inbuf, &r.index );
-            tr_peerIoReadUint32( msgs->io, inbuf, &r.offset );
-            tr_peerIoReadUint32( msgs->io, inbuf, &r.length );
-            reqListRemove( &msgs->clientAskedFor, &r );
-            break;
-        }
-
-        case BT_ALLOWED_FAST:
-        {
-            dbgmsg( msgs, "Got a BT_ALLOWED_FAST" );
-            tr_peerIoReadUint32( msgs->io, inbuf, &ui32 );
-            /* we don't do anything with this yet */
+            dbgmsg( msgs, "Got a BT_FEXT_REJECT" );
+            tr_peerIoReadUint32( msgs->peer->io, inbuf, &r.index );
+            tr_peerIoReadUint32( msgs->peer->io, inbuf, &r.offset );
+            tr_peerIoReadUint32( msgs->peer->io, inbuf, &r.length );
+            if( fext )
+                reqListRemove( &msgs->clientAskedFor, &r );
+            else {
+                fireError( msgs, EMSGSIZE );
+                return READ_ERR;
+            }
             break;
         }
@@ -1547 +1591 @@
         default:
             dbgmsg( msgs, "peer sent us an UNKNOWN: %d", (int)id );
-            tr_peerIoDrain( msgs->io, inbuf, msglen );
+            tr_peerIoDrain( msgs->peer->io, inbuf, msglen );
             break;
     }
@@ -1559 +1603 @@
 
 static void
-decrementDownloadedCount( tr_peermsgs * msgs,
-                          uint32_t      byteCount )
+decrementDownloadedCount( tr_peermsgs * msgs, uint32_t byteCount )
 {
     tr_torrent * tor = msgs->torrent;
@@ -1568 +1611 @@
 
 static void
-clientGotUnwantedBlock( tr_peermsgs *               msgs,
-                        const struct peer_request * req )
+clientGotUnwantedBlock( tr_peermsgs * msgs, const struct peer_request * req )
 {
     decrementDownloadedCount( msgs, req->length );
@@ -1575 +1617 @@
 
 static void
-addPeerToBlamefield( tr_peermsgs * msgs,
-                     uint32_t      index )
-{
-    if( !msgs->info->blame )
-        msgs->info->blame = tr_bitfieldNew( msgs->torrent->info.pieceCount );
-    tr_bitfieldAdd( msgs->info->blame, index );
+addPeerToBlamefield( tr_peermsgs * msgs, uint32_t index )
+{
+    if( !msgs->peer->blame )
+         msgs->peer->blame = tr_bitfieldNew( msgs->torrent->info.pieceCount );
+    tr_bitfieldAdd( msgs->peer->blame, index );
 }
 
@@ -1589 +1630 @@
                 const struct peer_request * req )
 {
-    int                    err;
-    tr_torrent *           tor = msgs->torrent;
+    int err;
+    tr_torrent * tor = msgs->torrent;
     const tr_block_index_t block = _tr_block( tor, req->index, req->offset );
 
@@ -1596 +1637 @@
     assert( req );
 
-    if( req->length != tr_torBlockCountBytes( msgs->torrent, block ) )
-    {
+    if( req->length != tr_torBlockCountBytes( msgs->torrent, block ) ) {
         dbgmsg( msgs, "wrong block size -- expected %u, got %d",
                 tr_torBlockCountBytes( msgs->torrent, block ), req->length );
@@ -1604 +1644 @@
 
     /* save the block */
-    dbgmsg( msgs, "got block %u:%u->%u", req->index, req->offset,
-            req->length );
+    dbgmsg( msgs, "got block %u:%u->%u", req->index, req->offset, req->length );
 
     /**
@@ -1611 +1650 @@
     **/
 
-    if( !reqListRemove( &msgs->clientAskedFor, req ) )
-    {
+    if( !reqListRemove( &msgs->clientAskedFor, req ) ) {
         clientGotUnwantedBlock( msgs, req );
         dbgmsg( msgs, "we didn't ask for this message..." );
@@ -1625 +1663 @@
     **/
 
-    if( tr_cpBlockIsComplete( tor->completion, block ) )
-    {
+    if( tr_cpBlockIsComplete( tor->completion, block ) ) {
         dbgmsg( msgs, "we have this block already..." );
         clientGotUnwantedBlock( msgs, req );
@@ -1655 +1692 @@
 
 static ReadState
-canRead( struct tr_iobuf * iobuf, void * vmsgs, size_t * piece )
+canRead( tr_peerIo * io, void * vmsgs, size_t * piece )
 {
     ReadState         ret;
     tr_peermsgs *     msgs = vmsgs;
-    struct evbuffer * in = tr_iobuf_input( iobuf );
+    struct evbuffer * in = tr_peerIoGetReadBuffer( io );
     const size_t      inlen = EVBUFFER_LENGTH( in );
 
@@ -1700 +1737 @@
 {
     tr_peermsgs * msgs = vmsgs;
-    const double rateToClient = tr_peerGetPieceSpeed( msgs->info, TR_PEER_TO_CLIENT );
-    const int estimatedBlocksInNext30Seconds =
-                  ( rateToClient * 30 * 1024 ) / msgs->torrent->blockSize;
-    msgs->minActiveRequests = 8;
-    msgs->maxActiveRequests = msgs->minActiveRequests + estimatedBlocksInNext30Seconds;
+    const double rateToClient = tr_peerGetPieceSpeed( msgs->peer, TR_PEER_TO_CLIENT );
+    const int seconds = 10;
+    const int floor = 8;
+    const int estimatedBlocksInPeriod = ( rateToClient * seconds * 1024 ) / msgs->torrent->blockSize;
+
+    msgs->maxActiveRequests = floor + estimatedBlocksInPeriod;
+
     if( msgs->reqq > 0 )
         msgs->maxActiveRequests = MIN( msgs->maxActiveRequests, msgs->reqq );
+
     return TRUE;
-}
-
-static int
-popNextRequest( tr_peermsgs *         msgs,
-                struct peer_request * setme )
-{
-    return reqListPop( &msgs->peerAskedForFast, setme )
-        || reqListPop( &msgs->peerAskedFor, setme );
 }
 
@@ -1723 +1755 @@
     size_t bytesWritten = 0;
     struct peer_request req;
-    const int haveMessages = EVBUFFER_LENGTH( msgs->outMessages ) != 0;
+    const tr_bool haveMessages = EVBUFFER_LENGTH( msgs->outMessages ) != 0;
+    const tr_bool fext = tr_peerIoSupportsFEXT( msgs->peer->io );
 
     /**
@@ -1738 +1771 @@
         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 );
+        dbgmsg( msgs, "flushing outMessages... to %p (length is %zu)", msgs->peer->io, len );
+        tr_peerIoWriteBuf( msgs->peer->io, msgs->outMessages, FALSE );
         msgs->clientSentAnythingAt = now;
         msgs->outMessagesBatchedAt = 0;
@@ -1750 +1783 @@
     **/
 
-    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 );
-            bytesWritten = 0;
-            msgs = NULL;
-        } 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;
+    if( ( tr_peerIoGetWriteBufferSpace( msgs->peer->io ) >= msgs->torrent->blockSize )
+        && popNextRequest( msgs, &req ) )
+    {
+        if( 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 );
+                bytesWritten = 0;
+                msgs = NULL;
+            } else {
+                tr_peerIo * io = msgs->peer->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 );
         }
-        tr_free( buf );
+        else if( fext ) /* peer needs a reject message */
+        {
+            protocolSendReject( msgs, &req );
+        }
     }
 
@@ -1788 +1827 @@
     {
         dbgmsg( msgs, "sending a keepalive message" );
-        tr_peerIoWriteUint32( msgs->io, msgs->outMessages, 0 );
+        tr_peerIoWriteUint32( msgs->peer->io, msgs->outMessages, 0 );
         pokeBatchPeriod( msgs, IMMEDIATE_PRIORITY_INTERVAL_SECS );
     }
@@ -1821 +1860 @@
 
 static void
-gotError( struct tr_iobuf  * iobuf UNUSED,
-          short              what,
-          void             * vmsgs )
+gotError( tr_peerIo  * io UNUSED,
+          short        what,
+          void       * vmsgs )
 {
     if( what & EVBUFFER_TIMEOUT )
@@ -1872 +1911 @@
     }
 
-    tr_peerIoWriteUint32( msgs->io, out,
+    tr_peerIoWriteUint32( msgs->peer->io, out,
                           sizeof( uint8_t ) + field->byteCount );
-    tr_peerIoWriteUint8 ( msgs->io, out, BT_BITFIELD );
-    tr_peerIoWriteBytes ( msgs->io, out, field->bits, field->byteCount );
-    dbgmsg( msgs, "sending bitfield... outMessage size is now %d",
-           (int)EVBUFFER_LENGTH( out ) );
+    tr_peerIoWriteUint8 ( msgs->peer->io, out, BT_BITFIELD );
+    tr_peerIoWriteBytes ( msgs->peer->io, out, field->bits, field->byteCount );
+    dbgmsg( msgs, "sending bitfield... outMessage size is now %zu",
+            EVBUFFER_LENGTH( out ) );
     pokeBatchPeriod( msgs, IMMEDIATE_PRIORITY_INTERVAL_SECS );
 
@@ -1884 +1923 @@
 
     tr_bitfieldFree( field );
+}
+
+static void
+tellPeerWhatWeHave( tr_peermsgs * msgs )
+{
+    const tr_bool fext = tr_peerIoSupportsFEXT( msgs->peer->io );
+
+    if( fext && ( tr_cpGetStatus( msgs->torrent->completion ) == TR_CP_COMPLETE ) )
+    {
+        protocolSendHaveAll( msgs );
+    }
+    else if( fext && ( tr_cpHaveValid( msgs->torrent->completion ) == 0 ) )
+    {
+        protocolSendHaveNone( msgs );
+    }
+    else
+    {
+        sendBitfield( msgs );
+    }
 }
 
@@ -1995 +2053 @@
             tmp = walk = tr_new( uint8_t, diffs.addedCount * 6 );
             for( i = 0; i < diffs.addedCount; ++i ) {
-                memcpy( walk, &diffs.added[i].in_addr, 4 ); walk += 4;
+                memcpy( walk, &diffs.added[i].addr, 4 ); walk += 4;
                 memcpy( walk, &diffs.added[i].port, 2 ); walk += 2;
             }
@@ -2013 +2071 @@
             tmp = walk = tr_new( uint8_t, diffs.droppedCount * 6 );
             for( i = 0; i < diffs.droppedCount; ++i ) {
-                memcpy( walk, &diffs.dropped[i].in_addr, 4 ); walk += 4;
+                memcpy( walk, &diffs.dropped[i].addr, 4 ); walk += 4;
                 memcpy( walk, &diffs.dropped[i].port, 2 ); walk += 2;
             }
@@ -2022 +2080 @@
             /* write the pex message */
             benc = tr_bencSave( &val, &bencLen );
-            tr_peerIoWriteUint32( msgs->io, out, 2 * sizeof( uint8_t ) + bencLen );
-            tr_peerIoWriteUint8 ( msgs->io, out, BT_LTEP );
-            tr_peerIoWriteUint8 ( msgs->io, out, msgs->ut_pex_id );
-            tr_peerIoWriteBytes ( msgs->io, out, benc, bencLen );
+            tr_peerIoWriteUint32( msgs->peer->io, out, 2 * sizeof( uint8_t ) + bencLen );
+            tr_peerIoWriteUint8 ( msgs->peer->io, out, BT_LTEP );
+            tr_peerIoWriteUint8 ( msgs->peer->io, out, msgs->ut_pex_id );
+            tr_peerIoWriteBytes ( msgs->peer->io, out, benc, bencLen );
             pokeBatchPeriod( msgs, HIGH_PRIORITY_INTERVAL_SECS );
             dbgmsg( msgs, "sending a pex message; outMessage size is now %zu", EVBUFFER_LENGTH( out ) );
@@ -2055 +2113 @@
 tr_peermsgs*
 tr_peerMsgsNew( struct tr_torrent * torrent,
-                struct tr_peer *    info,
+                struct tr_peer    * peer,
                 tr_delivery_func    func,
-                void *              userData,
-                tr_publisher_tag *  setme )
+                void              * userData,
+                tr_publisher_tag  * setme )
 {
     tr_peermsgs * m;
 
-    assert( info );
-    assert( info->io );
+    assert( peer );
+    assert( peer->io );
 
     m = tr_new0( tr_peermsgs, 1 );
     m->publisher = tr_publisherNew( );
-    m->info = info;
+    m->peer = peer;
     m->session = torrent->session;
     m->torrent = torrent;
-    m->io = info->io;
-    m->info->clientIsChoked = 1;
-    m->info->peerIsChoked = 1;
-    m->info->clientIsInterested = 0;
-    m->info->peerIsInterested = 0;
-    m->info->have = tr_bitfieldNew( torrent->info.pieceCount );
+    m->peer->clientIsChoked = 1;
+    m->peer->peerIsChoked = 1;
+    m->peer->clientIsInterested = 0;
+    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 );
@@ -2082 +2139 @@
     m->outMessagesBatchPeriod = LOW_PRIORITY_INTERVAL_SECS;
     m->incoming.block = evbuffer_new( );
-    m->peerAllowedPieces = NULL;
     m->peerAskedFor = REQUEST_LIST_INIT;
-    m->peerAskedForFast = REQUEST_LIST_INIT;
     m->clientAskedFor = REQUEST_LIST_INIT;
     m->clientWillAskFor = REQUEST_LIST_INIT;
+    peer->msgs = m;
+
     *setme = tr_publisherSubscribe( m->publisher, func, userData );
 
-    if( tr_peerIoSupportsLTEP( m->io ) )
+    if( tr_peerIoSupportsLTEP( peer->io ) )
         sendLtepHandshake( m );
 
-    sendBitfield( m );
-
-    tr_peerIoSetTimeoutSecs( m->io, 150 ); /* timeout after N seconds of
-                                             inactivity */
-    tr_peerIoSetIOFuncs( m->io, canRead, didWrite, gotError, m );
+    tellPeerWhatWeHave( m );
+
+    tr_peerIoSetIOFuncs( m->peer->io, canRead, didWrite, gotError, m );
     ratePulse( m );
 
@@ -2111 +2166 @@
         reqListClear( &msgs->clientWillAskFor );
         reqListClear( &msgs->clientAskedFor );
-        reqListClear( &msgs->peerAskedForFast );
         reqListClear( &msgs->peerAskedFor );
-        tr_bitfieldFree( msgs->peerAllowedPieces );
+
         evbuffer_free( msgs->incoming.block );