Changeset 2978

Timestamp:

Sep 6, 2007, 12:33:33 AM (16 years ago)

Author:

charles

Message:

still way too buggy, but I can now seed at full speed using < 2% of the CPU. =)

Location:

branches/encryption/libtransmission

Files:

• fastresume.c (modified) (6 diffs)
• handshake.c (modified) (6 diffs)
• peer-io.c (modified) (1 diff)
• peer-mgr-private.h (modified) (2 diffs)
• peer-mgr.c (modified) (8 diffs)
• peer-mgr.h (modified) (4 diffs)
• peer-msgs.c (modified) (11 diffs)
• utils.c (modified) (1 diff)
• utils.h (modified) (1 diff)

branches/encryption/libtransmission/fastresume.c

@@ -75 +75 @@
 
     /* IPs and ports of connectable peers */
-    FR_ID_PEERS = 4,
+    FR_ID_PEERS_OLD = 4,
 
     /* progress data:
@@ -101 +101 @@
 
     /* number of corrupt bytes downloaded */
-    FR_ID_CORRUPT = 10
+    FR_ID_CORRUPT = 10,
+
+    /* IPs and ports of connectable peers */
+    FR_ID_PEERS = 11
 };
 
@@ -284 +287 @@
 
     /* Write download and upload totals */
+
     total = tor->downloadedCur + tor->downloadedPrev;
     fastResumeWriteData( FR_ID_DOWNLOADED, &total, 8, 1, file );
+
     total = tor->uploadedCur + tor->uploadedPrev;
     fastResumeWriteData( FR_ID_UPLOADED, &total, 8, 1, file );
+
     total = tor->corruptCur + tor->corruptPrev;
     fastResumeWriteData( FR_ID_CORRUPT, &total, 8, 1, file );
@@ -293 +299 @@
     if( !( TR_FLAG_PRIVATE & tor->info.flags ) )
     {
-        /* Write IPs and ports of connectable peers, if any */
-        uint8_t * buf = NULL;
-        const int size = tr_peerMgrGetPeers( tor->handle->peerMgr,
-                                             tor->info.hash, &buf );
-        if( size > 0 )
-        {
-            fastResumeWriteData( FR_ID_PEERS, buf, size, 1, file );
-            free( buf );
-        }
+        tr_pex * pex;
+        const int count = tr_peerMgrGetPeers( tor->handle->peerMgr,
+                                              tor->info.hash,
+                                              &pex );
+        if( count > 0 )
+            fastResumeWriteData( FR_ID_PEERS, pex, sizeof(tr_pex), count, file );
+        tr_free( pex );
     }
 
@@ -638 +642 @@
                 break;
 
-            case FR_ID_PEERS:
+            case FR_ID_PEERS_OLD:
                 if( !( TR_FLAG_PRIVATE & tor->info.flags ) )
                 {
@@ -658 +662 @@
                     ret |= TR_FR_PEERS;
                 }
+
+            case FR_ID_PEERS:
+                if( !( TR_FLAG_PRIVATE & tor->info.flags ) )
+                {
+                    const int count = len / sizeof(tr_pex);
+                    tr_pex * pex = tr_new0( tr_pex, count );
+                    if( 1 != fread( pex, sizeof(tr_pex), count, file ) )
+                    {
+                        free( pex );
+                        fclose( file );
+                        return ret;
+                    }
+
+                    tr_peerMgrAddPex( tor->handle->peerMgr,
+                                      tor->info.hash,
+                                      TR_PEER_FROM_CACHE,
+                                      pex, count );
+
+                    tr_dbg( "found %i peers in resume file", len/6 );
+                    free( pex );
+                    ret |= TR_FR_PEERS;
+                }
                 continue;
 

branches/encryption/libtransmission/handshake.c

@@ -12 +12 @@
 
 #include <assert.h>
+#include <errno.h>
 #include <inttypes.h>
 #include <limits.h> /* UCHAR_MAX */
@@ -32 +33 @@
 
 /* enable Azureus messaging protocol */
-#define ENABLE_AZMP
+//#define ENABLE_AZMP
 
 /***
@@ -433 +434 @@
     int isEncrypted;
     const uint8_t * secret;
-    const size_t needlen = KEY_LEN;
     uint8_t yb[KEY_LEN];
     struct evbuffer * outbuf;
+    size_t needlen = HANDSHAKE_NAME_LEN;
 
     if( EVBUFFER_LENGTH(inbuf) < needlen )
         return READ_MORE;
+fprintf( stderr, "%*.*s\n", HANDSHAKE_NAME_LEN, HANDSHAKE_NAME_LEN, EVBUFFER_DATA(inbuf) );
 
     isEncrypted = memcmp( EVBUFFER_DATA(inbuf), HANDSHAKE_NAME, HANDSHAKE_NAME_LEN );
+    if( isEncrypted ) {
+        needlen = KEY_LEN;
+        if( EVBUFFER_LENGTH(inbuf) < needlen )
+            return READ_MORE;
+    }
+
     fprintf( stderr, "got a %s handshake\n", (isEncrypted ? "encrypted" : "plaintext") );
     tr_peerIoSetEncryption( handshake->io, isEncrypted
@@ -671 +679 @@
     /* peer id */
     tr_peerIoReadBytes( handshake->io, inbuf, peer_id, sizeof(peer_id) );
-    fprintf( stderr, "peer_id: " );
-    for( i=0; i<20; ++i ) fprintf( stderr, "[%c]", peer_id[i] );
-    fprintf( stderr, "\n" );
+//    fprintf( stderr, "peer_id: " );
+//    for( i=0; i<20; ++i ) fprintf( stderr, "[%c]", peer_id[i] );
+//    fprintf( stderr, "\n" );
     tr_peerIoSetPeersId( handshake->io, peer_id );
     bytesRead += sizeof(peer_id);
@@ -801 +809 @@
 {
     tr_handshake * handshake = (tr_handshake *) arg;
-fprintf( stderr, "handshake %p: got error; what==%hd\n", handshake, what );
+fprintf( stderr, "handshake %p: got error [%s]; what==%hd... state was [%s]\n", handshake, strerror(errno), what, getStateName(handshake->state) );
 
 
@@ -807 +815 @@
      * have encountered a peer that doesn't do encryption... reconnect and
      * try a plaintext handshake */
-    if(    ( handshake->state == AWAITING_YB )
+    if(    ( ( handshake->state == AWAITING_YB ) || ( handshake->state == AWAITING_VC ) )
         && ( handshake->encryptionPreference != HANDSHAKE_ENCRYPTION_REQUIRED )
         && ( !tr_peerIoReconnect( handshake->io ) ) )
     {
+fprintf( stderr, "handshake %p trying again in plaintext...\n", handshake );
         handshake->encryptionPreference = HANDSHAKE_PLAINTEXT_REQUIRED;
         sendPlaintextHandshake( handshake );

branches/encryption/libtransmission/peer-io.c

@@ -226 +226 @@
     assert( !tr_peerIoIsIncoming( io ) );
 
+fprintf( stderr, "tr_peerIoReconnect: io %p\n", io );
+
     if( io->socket >= 0 )
         tr_netClose( io->socket );
 
-    io->socket = tr_netOpenTCP( &io->in_addr,
-                                        io->port, 0 );
-
-    return io->socket >= 0 ? 0 : -1;
+    io->socket = tr_netOpenTCP( &io->in_addr, io->port, 0 );
+
+fprintf( stderr, "tr_peerIoReconnect: io->socket is %d\n", io->socket );
+  
+    if( io->socket >= 0 )
+    {
+        bufferevent_free( io->bufev );
+
+        io->bufev = bufferevent_new( io->socket,
+                                     canReadWrapper,
+                                     didWriteWrapper,
+                                     gotErrorWrapper,
+                                     io );
+        bufferevent_enable( io->bufev, EV_READ|EV_WRITE );
+
+        return 0;
+    }
+ 
+    return -1;
 }
 

branches/encryption/libtransmission/peer-mgr-private.h

@@ -16 +16 @@
 #include <inttypes.h> /* uint16_t */
 #include <arpa/inet.h> /* struct in_addr */
+
 struct tr_bitfield;
 struct tr_peerIo;
@@ -44 +45 @@
 
     struct tr_peermsgs * msgs;
+
+    struct tr_pex * lastPex;
+    int lastPexCount;
 }
 tr_peer;

branches/encryption/libtransmission/peer-mgr.c

@@ -226 +226 @@
         peer->port = port;
         peer->msgs = tr_peerMsgsNew( t->tor, peer );
+        chokePulse( t );
     }
 }
@@ -240 +241 @@
                      myHandshakeDoneCB,
                      manager );
+}
+
+static void
+maybeConnect( tr_peerMgr * manager, Torrent * t, tr_peer * peer )
+{
+    if( tr_peerMgrIsAcceptingConnections( manager ) )
+    {
+        fprintf( stderr, "peer-mgr: torrent [%s] is handshaking with a new peer %08x:%04x\n",
+                 t->tor->info.name,
+                 (uint32_t) peer->in_addr.s_addr, peer->port );
+
+        peer->io = tr_peerIoNewOutgoing( manager->handle, &peer->in_addr, peer->port, t->hash );
+
+        tr_handshakeAdd( peer->io, HANDSHAKE_ENCRYPTION_PREFERRED,
+                         myHandshakeDoneCB, manager );
+    }
+}
+
+void
+tr_peerMgrAddPex( tr_peerMgr     * manager,
+                  const uint8_t  * torrentHash,
+                  int              from,
+                  const tr_pex   * pex,
+                  int              pexCount )
+{
+    int i;
+    const tr_pex * walk = pex;
+    Torrent * t = getExistingTorrent( manager, torrentHash );
+    for( i=0; i<pexCount; ++i )
+    {
+        tr_peer * peer = getPeer( t, &walk->in_addr );
+        peer->port = walk->port;
+        peer->from = from;
+        maybeConnect( manager, t, peer );
+    }
 }
 
@@ -252 +288 @@
     const uint8_t * walk = peerCompact;
     Torrent * t = getExistingTorrent( manager, torrentHash );
-
     for( i=0; i<peerCount; ++i )
     {
@@ -258 +293 @@
         struct in_addr addr;
         uint16_t port;
-
         memcpy( &addr, walk, 4 ); walk += 4;
         memcpy( &port, walk, 2 ); walk += 2;
-
         peer = getPeer( t, &addr );
         peer->port = port;
         peer->from = from;
-
-        if( tr_peerMgrIsAcceptingConnections( manager ) )
-        {
-            fprintf( stderr, "peer-mgr: torrent [%s] is handshaking with a new peer...\n", t->tor->info.name );
-
-            peer->io = tr_peerIoNewOutgoing( manager->handle, &addr, port, t->hash );
-
-            tr_handshakeAdd( peer->io, HANDSHAKE_ENCRYPTION_PREFERRED,
-                             myHandshakeDoneCB, manager );
-        }
+        maybeConnect( manager, t, peer );
     }
 }
@@ -298 +322 @@
 
 int
+tr_pexCompare( const void * va, const void * vb )
+{
+    const tr_pex * a = (const tr_pex *) va;
+    const tr_pex * b = (const tr_pex *) 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;
+    return 0;
+}
+
+int tr_pexCompare( const void * a, const void * b );
+
+
+int
 tr_peerMgrGetPeers( tr_peerMgr      * manager,
                     const uint8_t   * torrentHash,
-                    uint8_t        ** setme_compact )
+                    tr_pex         ** setme_pex )
 {
     const Torrent * t = getExistingTorrent( (tr_peerMgr*)manager, torrentHash );
     int i, peerCount;
     const tr_peer ** peers = (const tr_peer **) tr_ptrArrayPeek( t->peers, &peerCount );
-    uint8_t * ret = tr_new( uint8_t, peerCount * 6 );
-    uint8_t * walk = ret;
-
-    for( i=0; i<peerCount; ++i )
-    {
-        memcpy( walk, &peers[i]->in_addr, sizeof( struct in_addr ) );
-        walk += sizeof( struct in_addr );
-        memcpy( walk, &peers[i]->port, sizeof( uint16_t ) );
-        walk += sizeof( uint16_t );
-    }
-
-    assert( ( walk - ret ) == peerCount * 6 );
-    *setme_compact = ret;
+    tr_pex * pex = tr_new( tr_pex, peerCount );
+    tr_pex * walk = pex;
+
+    for( i=0; i<peerCount; ++i, ++walk )
+    {
+        walk->in_addr = peers[i]->in_addr;
+        walk->port = peers[i]->port;
+        walk->flags = '\0'; /* FIXME */
+    }
+
+    assert( ( walk - pex ) == peerCount );
+    qsort( pex, peerCount, sizeof(tr_pex), tr_pexCompare );
+    *setme_pex = pex;
     return peerCount;
 }
@@ -524 +563 @@
 
 static int
-chokePulse( void * vtorrent UNUSED )
+chokePulse( void * vtorrent )
 {
     Torrent * t = (Torrent *) vtorrent;
-    int i, size, optimistic;
+    int i, size, unchoked;
     const int done = tr_cpGetStatus( t->tor->completion ) != TR_CP_INCOMPLETE;
     tr_peer ** peers = (tr_peer **) tr_ptrArrayPeek( t->peers, &size );
@@ -549 +588 @@
     qsort( data, size, sizeof(ChokeData), compareChokeByDownloader );
     bestDownloaderRate = data[0].rate;
-    for( i=0; i<size && i<NUM_DOWNLOADERS_TO_UNCHOKE; ++i )
-        tr_peerMsgsSetChoke( data[i].peer->msgs, FALSE );
+    for( i=unchoked=0; i<size && unchoked<NUM_DOWNLOADERS_TO_UNCHOKE; ++i ) {
+        if( data[i].peer->msgs != NULL ) {
+            tr_peerMsgsSetChoke( data[i].peer->msgs, FALSE );
+            ++unchoked;
+        }
+    }
     memmove( data, data+i, sizeof(ChokeData)*(size-i) );
     size -= i;
@@ -562 +605 @@
 
     /* of those remaining, optimistically unchoke one; choke the rest */
-    optimistic = tr_rand( size );
-    for( i=0; i<size; ++i )
-        tr_peerMsgsSetChoke( data[i].peer->msgs, i!=optimistic );
+    if( size > 0 ) {
+        const int optimistic = tr_rand( size );
+        for( i=0; i<size; ++i )
+            tr_peerMsgsSetChoke( data[i].peer->msgs, i!=optimistic );
+    }
 
     /* cleanup */

branches/encryption/libtransmission/peer-mgr.h

@@ -14 +14 @@
 #define TR_PEER_MGR_H
 
+#include <inttypes.h> /* uint16_t */
+#include <arpa/inet.h> /* struct in_addr */
+
 struct in_addr;
 struct tr_handle;
@@ -19 +22 @@
 struct tr_torrent;
 typedef struct tr_peerMgr tr_peerMgr;
+
+typedef struct tr_pex
+{
+    struct in_addr in_addr;
+    uint16_t port;
+    uint8_t flags;
+}
+tr_pex;
+
+int tr_pexCompare( const void * a, const void * b );
 
 tr_peerMgr* tr_peerMgrNew( struct tr_handle * );
@@ -36 +49 @@
                          int              peerCount );
 
+void tr_peerMgrAddPex( tr_peerMgr     * manager,
+                       const uint8_t  * torrentHash,
+                       int              from,
+                       const tr_pex   * pex,
+                       int              pexCount );
+
 void tr_peerMgrSetBlame( tr_peerMgr     * manager,
                          const uint8_t  * torrentHash,
@@ -43 +62 @@
 int tr_peerMgrGetPeers( tr_peerMgr      * manager,
                         const uint8_t   * torrentHash,
-                        uint8_t        ** setme_compact );
+                        tr_pex         ** setme_pex );
 
 void tr_peerMgrStartTorrent( tr_peerMgr     * manager,

branches/encryption/libtransmission/peer-msgs.c

@@ -39 +39 @@
 **/
 
+#define MINUTES_TO_MSEC(N) ((N) * 60 * 1000)
+
+/* pex attempts are made this frequently */
+#define PEX_INTERVAL (MINUTES_TO_MSEC(1))
+
 enum
 {
@@ -110 +115 @@
 
     tr_timer_tag pulseTag;
+    tr_timer_tag pexTag;
 
     unsigned int  notListening        : 1;
@@ -121 +127 @@
     uint64_t gotKeepAliveTime;
 
-    uint16_t ut_pex;
+    uint8_t ut_pex;
     uint16_t listeningPort;
+
+    tr_pex * pex;
+    int pexCount;
 };
 
@@ -186 +195 @@
 tr_peerMsgsSetChoke( tr_peermsgs * peer, int choke )
 {
+    assert( peer != NULL );
+
     if( peer->info->peerIsChoked != !!choke )
     {
@@ -228 +239 @@
         sub = tr_bencDictFind( sub, "ut_pex" );
         if( tr_bencIsInt( sub ) ) {
-            peer->ut_pex = sub->val.i;
+            peer->ut_pex = (uint8_t) sub->val.i;
             fprintf( stderr, "peer->ut_pex is %d\n", peer->ut_pex );
         }
@@ -598 +609 @@
     size_t len;
 
+fprintf( stderr, "peer %p pulse... notlistening %d, outblock size: %d, outmessages size %d, peerAskedFor %p\n",
+         vpeer,
+         (int)peer->notListening,
+         (int)EVBUFFER_LENGTH( peer->outBlock ),
+         (int)EVBUFFER_LENGTH( peer->outMessages ),
+         peer->peerAskedFor );
+
     /* if we froze out a downloaded block because of speed limits,
        start listening to the peer again */
@@ -609 +627 @@
     if(( len = EVBUFFER_LENGTH( peer->outBlock ) ))
     {
+fprintf( stderr, "peer %p needing to upload... canUpload %d\n", peer, canUpload(peer) );
         if( canUpload( peer ) )
         {
             const size_t outlen = MIN( len, 2048 );
+fprintf( stderr, "peer %p writing %d bytes...\n", peer, (int)outlen );
             tr_peerIoWrite( peer->io, EVBUFFER_DATA(peer->outBlock), outlen );
             evbuffer_drain( peer->outBlock, outlen );
+
+            peer->torrent->uploadedCur += outlen;
+            tr_rcTransferred( peer->torrent->upload, outlen );
+            tr_rcTransferred( peer->handle->upload, outlen );
         }
     }
@@ -628 +652 @@
         const uint8_t msgid = BT_PIECE;
         const uint32_t msglen = sizeof(uint8_t) + sizeof(uint32_t)*2 + req->length;
+fprintf( stderr, "peer %p starting to upload a block...\n", peer );
         tr_ioRead( peer->torrent, req->index, req->offset, req->length, tmp );
         tr_peerIoWriteUint32( peer->io, peer->outBlock, msglen );
@@ -667 +692 @@
     tr_peerIoWriteBytes( peer->io, peer->outMessages, bitfield->bits, bitfield->len );
 }
+
+/**
+***
+**/
+
+#define MAX_DIFFS 50
+
+typedef struct
+{
+    tr_pex * added;
+    tr_pex * dropped;
+    tr_pex * elements;
+    int addedCount;
+    int droppedCount;
+    int elementCount;
+    int diffCount;
+}
+PexDiffs;
+
+static void pexAddedCb( void * vpex, void * userData )
+{
+    PexDiffs * diffs = (PexDiffs *) userData;
+    tr_pex * pex = (tr_pex *) vpex;
+    if( diffs->diffCount < MAX_DIFFS )
+    {
+        diffs->diffCount++;
+        diffs->added[diffs->addedCount++] = *pex;
+        diffs->elements[diffs->elementCount++] = *pex;
+    }
+}
+
+static void pexRemovedCb( void * vpex, void * userData )
+{
+    PexDiffs * diffs = (PexDiffs *) userData;
+    tr_pex * pex = (tr_pex *) vpex;
+    if( diffs->diffCount < MAX_DIFFS )
+    {
+        diffs->diffCount++;
+        diffs->dropped[diffs->droppedCount++] = *pex;
+    }
+}
+
+static void pexElementCb( void * vpex, void * userData )
+{
+    PexDiffs * diffs = (PexDiffs *) userData;
+    tr_pex * pex = (tr_pex *) vpex;
+    if( diffs->diffCount < MAX_DIFFS )
+    {
+        diffs->diffCount++;
+        diffs->elements[diffs->elementCount++] = *pex;
+    }
+}
+
+static int
+pexPulse( void * vpeer )
+{
+    tr_peermsgs * peer = (tr_peermsgs *) vpeer;
+
+    if( peer->info->pexEnabled )
+    {
+        int i;
+        tr_pex * newPex = NULL;
+        const int newCount = tr_peerMgrGetPeers( peer->handle->peerMgr, peer->torrent->info.hash, &newPex );
+        PexDiffs diffs;
+        benc_val_t val, *added, *dropped, *flags;
+        uint8_t *tmp, *walk;
+        char * benc;
+        int bencLen;
+        const uint8_t bt_msgid = BT_LTEP;
+        const uint8_t ltep_msgid = peer->ut_pex;
+
+        /* build the diffs */
+        diffs.added = tr_new( tr_pex, newCount );
+        diffs.addedCount = 0;
+        diffs.dropped = tr_new( tr_pex, peer->pexCount );
+        diffs.droppedCount = 0;
+        diffs.elements = tr_new( tr_pex, newCount + peer->pexCount );
+        diffs.elementCount = 0;
+        diffs.diffCount = 0;
+        tr_set_compare( peer->pex, peer->pexCount,
+                        newPex, newCount,
+                        tr_pexCompare, sizeof(tr_pex),
+                        pexRemovedCb, pexAddedCb, pexElementCb, &diffs );
+        fprintf( stderr, "pex: old peer count %d, new peer count %d, added %d, removed %d\n", peer->pexCount, newCount, diffs.addedCount, diffs.droppedCount );
+
+        /* update peer */
+        tr_free( peer->pex );
+        peer->pex = diffs.elements;
+        peer->pexCount = diffs.elementCount;
+
+       
+        /* build the pex payload */
+        tr_bencInit( &val, TYPE_DICT );
+        tr_bencDictReserve( &val, 3 );
+
+        /* "added" */
+        added = tr_bencDictAdd( &val, "added" );
+        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].port, 2 ); walk += 2;
+        }
+        assert( ( walk - tmp ) == diffs.addedCount * 6 );
+        tr_bencInitStr( added, tmp, walk-tmp, FALSE );
+
+        /* "added.f" */
+        flags = tr_bencDictAdd( &val, "added.f" );
+        tmp = walk = tr_new( uint8_t, diffs.addedCount );
+        for( i=0; i<diffs.addedCount; ++i )
+            *walk++ = diffs.added[i].flags;
+        assert( ( walk - tmp ) == diffs.addedCount );
+        tr_bencInitStr( flags, tmp, walk-tmp, FALSE );
+
+        /* "dropped" */
+        dropped = tr_bencDictAdd( &val, "dropped" );
+        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].port, 2 ); walk += 2;
+        }
+        assert( ( walk - tmp ) == diffs.droppedCount * 6 );
+        tr_bencInitStr( dropped, tmp, walk-tmp, FALSE );
+
+        /* write the pex message */
+        benc = tr_bencSaveMalloc( &val, &bencLen );
+        tr_peerIoWriteUint32( peer->io, peer->outBlock, 1 + 1 + bencLen );
+        tr_peerIoWriteBytes ( peer->io, peer->outBlock, &bt_msgid, 1 );
+        tr_peerIoWriteBytes ( peer->io, peer->outBlock, &ltep_msgid, 1 );
+        tr_peerIoWriteBytes ( peer->io, peer->outBlock, benc, bencLen );
+
+        /* cleanup */
+        tr_free( benc );
+        tr_bencFree( &val );
+        tr_free( diffs.added );
+        tr_free( diffs.dropped );
+        tr_free( newPex );
+    }
+
+    return TRUE;
+}
+
+/**
+***
+**/
 
 tr_peermsgs*
@@ -686 +855 @@
     peer->info->peerIsInterested = 0;
     peer->info->have = tr_bitfieldNew( torrent->info.pieceCount );
-    peer->pulseTag = tr_timerNew( peer->handle, pulse, peer, NULL, 200 );
+    peer->pulseTag = tr_timerNew( peer->handle, pulse, peer, NULL, 500 );
+fprintf( stderr, "peer %p pulseTag %p\n", peer, peer->pulseTag );
+    peer->pexTag = tr_timerNew( peer->handle, pexPulse, peer, NULL, PEX_INTERVAL );
     peer->outMessages = evbuffer_new( );
     peer->outBlock = evbuffer_new( );
@@ -704 +875 @@
     if( p != NULL )
     {
+fprintf( stderr, "peer %p destroying its pulse tag\n", p );
         tr_timerFree( &p->pulseTag );
+        tr_timerFree( &p->pexTag );
         evbuffer_free( p->outMessages );
         evbuffer_free( p->outBlock );

branches/encryption/libtransmission/utils.c

@@ -186 +186 @@
 }
 
+/***
+****
+***/
+
+void
+tr_set_compare( const void * va, size_t aCount,
+                const void * vb, size_t bCount,
+                int compare( const void * a, const void * b ),
+                size_t elementSize,
+                tr_set_func in_a_cb,
+                tr_set_func in_b_cb,
+                tr_set_func in_both_cb,
+                void * userData )
+{
+    size_t ai, bi;
+    const uint8_t * a = (const uint8_t *) va;
+    const uint8_t * b = (const uint8_t *) vb;
+
+    for( ai=bi=0; ai<aCount && bi<bCount; )
+    {
+        if( ai==aCount )
+        {
+            (*in_b_cb)( (void*)b, userData );
+            b += elementSize;
+        }
+        else if ( bi==bCount )
+        {
+            (*in_a_cb)( (void*)a, userData );
+            a += elementSize;
+        }
+        else
+        {
+            const int val = (*compare)( a, b );
+
+            if( !val )
+            {
+                (*in_both_cb)( (void*)a, userData );
+                a += elementSize;
+                b += elementSize;
+            }
+            else if( val < 0 )
+            {
+                (*in_a_cb)( (void*)a, userData );
+                a += elementSize;
+            }
+            else if( val > 0 )
+            {
+                (*in_b_cb)( (void*)b, userData );
+                b += elementSize;
+            }
+        }
+    }
+}
+
+/***
+****
+***/
+
 
 #if 0

branches/encryption/libtransmission/utils.h

@@ -139 +139 @@
 ***/
 
+typedef void (tr_set_func)(void * element, void * userData );
+
+void tr_set_compare( const void * a, size_t aCount,
+                     const void * b, size_t bCount,
+                     int compare( const void * a, const void * b ),
+                     size_t elementSize,
+                     tr_set_func in_a_cb,
+                     tr_set_func in_b_cb,
+                     tr_set_func in_both_cb,
+                     void * userData );
+                    
+
+/***
+****
+***/
+
 struct tr_bitfield
 {