source: trunk/libtransmission/peer-io.c @ 7195

Last change on this file since 7195 was 7195, checked in by charles, 13 years ago

(libT) patch from jhujhiti to add ipv6 support.

  • Property svn:keywords set to Date Rev Author Id
File size: 17.6 KB
Line 
1/*
2 * This file Copyright (C) 2007-2008 Charles Kerr <charles@rebelbase.com>
3 *
4 * This file is licensed by the GPL version 2.  Works owned by the
5 * Transmission project are granted a special exemption to clause 2(b)
6 * so that the bulk of its code can remain under the MIT license.
7 * This exemption does not extend to derived works not owned by
8 * the Transmission project.
9 *
10 * $Id: peer-io.c 7195 2008-11-30 00:47:18Z charles $
11 */
12
13#include <assert.h>
14#include <limits.h> /* INT_MAX */
15#include <string.h>
16#include <stdio.h>
17#include <unistd.h>
18
19#ifdef WIN32
20 #include <winsock2.h>
21#else
22 #include <arpa/inet.h> /* inet_ntoa */
23#endif
24
25#include <event.h>
26
27#include "transmission.h"
28#include "bandwidth.h"
29#include "crypto.h"
30#include "iobuf.h"
31#include "list.h"
32#include "net.h"
33#include "peer-io.h"
34#include "trevent.h"
35#include "utils.h"
36
37#define MAGIC_NUMBER 206745
38#define IO_TIMEOUT_SECS 8
39
40static size_t
41getPacketOverhead( size_t d )
42{
43    /**
44     * http://sd.wareonearth.com/~phil/net/overhead/
45     *
46     * TCP over Ethernet:
47     * Assuming no header compression (e.g. not PPP)
48     * Add 20 IPv4 header or 40 IPv6 header (no options)
49     * Add 20 TCP header
50     * Add 12 bytes optional TCP timestamps
51     * Max TCP Payload data rates over ethernet are thus:
52     *  (1500-40)/(38+1500) = 94.9285 %  IPv4, minimal headers
53     *  (1500-52)/(38+1500) = 94.1482 %  IPv4, TCP timestamps
54     *  (1500-52)/(42+1500) = 93.9040 %  802.1q, IPv4, TCP timestamps
55     *  (1500-60)/(38+1500) = 93.6281 %  IPv6, minimal headers
56     *  (1500-72)/(38+1500) = 92.8479 %  IPv6, TCP timestamps
57     *  (1500-72)/(42+1500) = 92.6070 %  802.1q, IPv6, ICP timestamps
58     */
59    static const double assumed_payload_data_rate = 94.0;
60
61    return (size_t)( d * ( 100.0 / assumed_payload_data_rate ) - d );
62}
63
64/**
65***
66**/
67
68#define dbgmsg( io, ... ) \
69    do { \
70        if( tr_deepLoggingIsActive( ) ) \
71            tr_deepLog( __FILE__, __LINE__, tr_peerIoGetAddrStr( io ), __VA_ARGS__ ); \
72    } while( 0 )
73
74struct tr_datatype
75{
76    tr_bool  isPieceData;
77    size_t   length;
78};
79
80struct tr_peerIo
81{
82    tr_bool                  isEncrypted;
83    tr_bool                  isIncoming;
84    tr_bool                  peerIdIsSet;
85    tr_bool                  extendedProtocolSupported;
86
87    int                      magicNumber;
88
89    uint8_t                  encryptionMode;
90    uint8_t                  timeout;
91    uint16_t                 port;
92    int                      socket;
93
94    uint8_t                  peerId[20];
95    time_t                   timeCreated;
96
97    tr_session             * session;
98
99    tr_address               addr;
100    struct tr_iobuf        * iobuf;
101    tr_list                * output_datatypes; /* struct tr_datatype */
102
103    tr_can_read_cb           canRead;
104    tr_did_write_cb          didWrite;
105    tr_net_error_cb          gotError;
106    void *                   userData;
107
108    size_t                   bufferSize[2];
109
110    tr_bandwidth           * bandwidth;
111    tr_crypto              * crypto;
112};
113
114/***
115****
116***/
117
118static void
119didWriteWrapper( struct tr_iobuf  * iobuf,
120                 size_t             bytes_transferred,
121                 void             * vio )
122{
123    tr_peerIo *  io = vio;
124
125    while( bytes_transferred )
126    {
127        struct tr_datatype * next = io->output_datatypes->data;
128        const size_t payload = MIN( next->length, bytes_transferred );
129        const size_t overhead = getPacketOverhead( payload );
130
131        tr_bandwidthUsed( io->bandwidth, TR_UP, payload, next->isPieceData );
132
133        if( overhead > 0 )
134            tr_bandwidthUsed( io->bandwidth, TR_UP, overhead, FALSE );
135
136        if( io->didWrite )
137            io->didWrite( io, payload, next->isPieceData, io->userData );
138
139        bytes_transferred -= payload;
140        next->length -= payload;
141        if( !next->length )
142            tr_free( tr_list_pop_front( &io->output_datatypes ) );
143    }
144
145    if( EVBUFFER_LENGTH( tr_iobuf_output( iobuf ) ) )
146        tr_iobuf_enable( io->iobuf, EV_WRITE );
147}
148
149static void
150canReadWrapper( struct tr_iobuf  * iobuf,
151                size_t             bytes_transferred UNUSED,
152                void              * vio )
153{
154    int          done = 0;
155    int          err = 0;
156    tr_peerIo *  io = vio;
157    tr_session * session = io->session;
158
159    dbgmsg( io, "canRead" );
160
161    /* try to consume the input buffer */
162    if( io->canRead )
163    {
164        tr_globalLock( session );
165
166        while( !done && !err )
167        {
168            size_t piece = 0;
169            const size_t oldLen = EVBUFFER_LENGTH( tr_iobuf_input( iobuf ) );
170            const int ret = io->canRead( iobuf, io->userData, &piece );
171
172            if( ret != READ_ERR )
173            {
174                const size_t used = oldLen - EVBUFFER_LENGTH( tr_iobuf_input( iobuf ) );
175                if( piece )
176                    tr_bandwidthUsed( io->bandwidth, TR_DOWN, piece, TRUE );
177                if( used != piece )
178                    tr_bandwidthUsed( io->bandwidth, TR_DOWN, used - piece, FALSE );
179            }
180
181            switch( ret )
182            {
183                case READ_NOW:
184                    if( EVBUFFER_LENGTH( tr_iobuf_input( iobuf )))
185                        continue;
186                    done = 1;
187                    break;
188
189                case READ_LATER:
190                    done = 1;
191                    break;
192
193                case READ_ERR:
194                    err = 1;
195                    break;
196            }
197        }
198
199        tr_globalUnlock( session );
200    }
201}
202
203static void
204gotErrorWrapper( struct tr_iobuf  * iobuf,
205                 short              what,
206                 void             * userData )
207{
208    tr_peerIo * c = userData;
209
210    if( c->gotError )
211        c->gotError( iobuf, what, c->userData );
212}
213
214/**
215***
216**/
217
218static void
219bufevNew( tr_peerIo * io )
220{
221    io->iobuf = tr_iobuf_new( io->session,
222                              io->bandwidth,
223                              io->socket,
224                              EV_READ | EV_WRITE,
225                              canReadWrapper,
226                              didWriteWrapper,
227                              gotErrorWrapper,
228                              io );
229
230    tr_iobuf_settimeout( io->iobuf, io->timeout, io->timeout );
231}
232
233static int
234isPeerIo( const tr_peerIo * io )
235{
236    return ( io != NULL ) && ( io->magicNumber == MAGIC_NUMBER );
237}
238
239static tr_peerIo*
240tr_peerIoNew( tr_session       * session,
241              const tr_address * addr,
242              uint16_t           port,
243              const uint8_t *    torrentHash,
244              int                isIncoming,
245              int                socket )
246{
247    tr_peerIo * io;
248
249    if( socket >= 0 )
250        tr_netSetTOS( socket, session->peerSocketTOS );
251
252    io = tr_new0( tr_peerIo, 1 );
253    io->magicNumber = MAGIC_NUMBER;
254    io->crypto = tr_cryptoNew( torrentHash, isIncoming );
255    io->session = session;
256    io->addr = *addr;
257    io->port = port;
258    io->socket = socket;
259    io->isIncoming = isIncoming != 0;
260    io->timeout = IO_TIMEOUT_SECS;
261    io->timeCreated = time( NULL );
262    bufevNew( io );
263    tr_peerIoSetBandwidth( io, session->bandwidth );
264    return io;
265}
266
267tr_peerIo*
268tr_peerIoNewIncoming( tr_session       * session,
269                      const tr_address * addr,
270                      uint16_t           port,
271                      int                socket )
272{
273    assert( session );
274    assert( addr );
275    assert( socket >= 0 );
276
277    return tr_peerIoNew( session, addr, port,
278                         NULL, 1,
279                         socket );
280}
281
282tr_peerIo*
283tr_peerIoNewOutgoing( tr_session       * session,
284                      const tr_address * addr,
285                      int                port,
286                      const uint8_t    * torrentHash )
287{
288    int socket;
289
290    assert( session );
291    assert( addr );
292    assert( port >= 0 );
293    assert( torrentHash );
294
295    socket = tr_netOpenTCP( session, addr, port );
296
297    return socket < 0
298           ? NULL
299           : tr_peerIoNew( session, addr, port, torrentHash, 0, socket );
300}
301
302static void
303io_dtor( void * vio )
304{
305    tr_peerIo * io = vio;
306
307    tr_peerIoSetBandwidth( io, NULL );
308    tr_iobuf_free( io->iobuf );
309    tr_netClose( io->socket );
310    tr_cryptoFree( io->crypto );
311    tr_list_free( &io->output_datatypes, tr_free );
312
313    io->magicNumber = 0xDEAD;
314    tr_free( io );
315}
316
317void
318tr_peerIoFree( tr_peerIo * io )
319{
320    if( io )
321    {
322        io->canRead = NULL;
323        io->didWrite = NULL;
324        io->gotError = NULL;
325        tr_runInEventThread( io->session, io_dtor, io );
326    }
327}
328
329tr_session*
330tr_peerIoGetSession( tr_peerIo * io )
331{
332    assert( isPeerIo( io ) );
333    assert( io->session );
334
335    return io->session;
336}
337
338const tr_address*
339tr_peerIoGetAddress( const tr_peerIo * io,
340                           uint16_t  * port )
341{
342    assert( isPeerIo( io ) );
343
344    if( port )
345        *port = io->port;
346
347    return &io->addr;
348}
349
350const char*
351tr_peerIoAddrStr( const tr_address * addr,
352                  uint16_t           port )
353{
354    static char buf[512];
355
356    if( addr->type == TR_AF_INET )
357        tr_snprintf( buf, sizeof( buf ), "%s:%u", tr_ntop_non_ts( addr ),
358                    ntohs( port ) );
359    else
360        tr_snprintf( buf, sizeof( buf ), "[%s]:%u", tr_ntop_non_ts( addr ),
361                    ntohs( port ) );
362    return buf;
363}
364
365const char*
366tr_peerIoGetAddrStr( const tr_peerIo * io )
367{
368    return tr_peerIoAddrStr( &io->addr, io->port );
369}
370
371static void
372tr_peerIoTryRead( tr_peerIo * io )
373{
374    if( EVBUFFER_LENGTH( tr_iobuf_input( io->iobuf )))
375        (*canReadWrapper)( io->iobuf, ~0, io );
376}
377
378void
379tr_peerIoSetIOFuncs( tr_peerIo *     io,
380                     tr_can_read_cb  readcb,
381                     tr_did_write_cb writecb,
382                     tr_net_error_cb errcb,
383                     void *          userData )
384{
385    io->canRead = readcb;
386    io->didWrite = writecb;
387    io->gotError = errcb;
388    io->userData = userData;
389
390    tr_peerIoTryRead( io );
391}
392
393int
394tr_peerIoIsIncoming( const tr_peerIo * c )
395{
396    return c->isIncoming ? 1 : 0;
397}
398
399int
400tr_peerIoReconnect( tr_peerIo * io )
401{
402    assert( !tr_peerIoIsIncoming( io ) );
403
404    if( io->socket >= 0 )
405        tr_netClose( io->socket );
406
407    io->socket = tr_netOpenTCP( io->session, &io->addr, io->port );
408
409    if( io->socket >= 0 )
410    {
411        tr_bandwidth * bandwidth = io->bandwidth;
412        tr_peerIoSetBandwidth( io, NULL );
413
414        tr_netSetTOS( io->socket, io->session->peerSocketTOS );
415        tr_iobuf_free( io->iobuf );
416        bufevNew( io );
417
418        tr_peerIoSetBandwidth( io, bandwidth );
419        return 0;
420    }
421
422    return -1;
423}
424
425void
426tr_peerIoSetTimeoutSecs( tr_peerIo * io,
427                         int         secs )
428{
429    io->timeout = secs;
430    tr_iobuf_settimeout( io->iobuf, io->timeout, io->timeout );
431    tr_iobuf_enable( io->iobuf, EV_READ | EV_WRITE );
432}
433
434/**
435***
436**/
437
438void
439tr_peerIoSetTorrentHash( tr_peerIo *     io,
440                         const uint8_t * hash )
441{
442    assert( isPeerIo( io ) );
443
444    tr_cryptoSetTorrentHash( io->crypto, hash );
445}
446
447const uint8_t*
448tr_peerIoGetTorrentHash( tr_peerIo * io )
449{
450    assert( isPeerIo( io ) );
451    assert( io->crypto );
452
453    return tr_cryptoGetTorrentHash( io->crypto );
454}
455
456int
457tr_peerIoHasTorrentHash( const tr_peerIo * io )
458{
459    assert( isPeerIo( io ) );
460    assert( io->crypto );
461
462    return tr_cryptoHasTorrentHash( io->crypto );
463}
464
465/**
466***
467**/
468
469void
470tr_peerIoSetPeersId( tr_peerIo *     io,
471                     const uint8_t * peer_id )
472{
473    assert( isPeerIo( io ) );
474
475    if( ( io->peerIdIsSet = peer_id != NULL ) )
476        memcpy( io->peerId, peer_id, 20 );
477    else
478        memset( io->peerId, 0, 20 );
479}
480
481const uint8_t*
482tr_peerIoGetPeersId( const tr_peerIo * io )
483{
484    assert( isPeerIo( io ) );
485    assert( io->peerIdIsSet );
486
487    return io->peerId;
488}
489
490/**
491***
492**/
493
494void
495tr_peerIoEnableLTEP( tr_peerIo * io,
496                     int         flag )
497{
498    assert( isPeerIo( io ) );
499    assert( flag == 0 || flag == 1 );
500
501    io->extendedProtocolSupported = flag;
502}
503
504int
505tr_peerIoSupportsLTEP( const tr_peerIo * io )
506{
507    assert( isPeerIo( io ) );
508
509    return io->extendedProtocolSupported;
510}
511
512/**
513***
514**/
515
516static size_t
517getDesiredOutputBufferSize( const tr_peerIo * io )
518{
519    /* this is all kind of arbitrary, but what seems to work well is
520     * being large enough to hold the next 15 seconds' worth of input,
521     * or two and a half blocks, whichever is bigger.
522     * It's okay to tweak this as needed */
523    const double maxBlockSize = 16 * 1024; /* 16 KiB is from BT spec */
524    const double currentSpeed = tr_bandwidthGetPieceSpeed( io->bandwidth, TR_UP );
525    const double period = 20; /* arbitrary */
526    return MAX( maxBlockSize*2.5, currentSpeed*1024*period );
527}
528
529size_t
530tr_peerIoGetWriteBufferSpace( const tr_peerIo * io )
531{
532    const size_t desiredLen = getDesiredOutputBufferSize( io );
533    const size_t currentLen = EVBUFFER_LENGTH( tr_iobuf_output( io->iobuf ) );
534    size_t freeSpace = 0;
535
536    if( desiredLen > currentLen )
537        freeSpace = desiredLen - currentLen;
538
539    return freeSpace;
540}
541
542void
543tr_peerIoSetBandwidth( tr_peerIo     * io,
544                       tr_bandwidth  * bandwidth )
545{
546    assert( isPeerIo( io ) );
547
548    if( io->bandwidth )
549        tr_bandwidthRemoveBuffer( io->bandwidth, io->iobuf );
550
551    io->bandwidth = bandwidth;
552    tr_iobuf_set_bandwidth( io->iobuf, bandwidth );
553
554    if( io->bandwidth )
555        tr_bandwidthAddBuffer( io->bandwidth, io->iobuf );
556
557    tr_iobuf_enable( io->iobuf, EV_READ | EV_WRITE );
558}
559
560/**
561***
562**/
563
564tr_crypto*
565tr_peerIoGetCrypto( tr_peerIo * c )
566{
567    return c->crypto;
568}
569
570void
571tr_peerIoSetEncryption( tr_peerIo * io,
572                        int         encryptionMode )
573{
574    assert( isPeerIo( io ) );
575    assert( encryptionMode == PEER_ENCRYPTION_NONE
576          || encryptionMode == PEER_ENCRYPTION_RC4 );
577
578    io->encryptionMode = encryptionMode;
579}
580
581int
582tr_peerIoIsEncrypted( const tr_peerIo * io )
583{
584    return io != NULL && io->encryptionMode == PEER_ENCRYPTION_RC4;
585}
586
587/**
588***
589**/
590
591void
592tr_peerIoWrite( tr_peerIo   * io,
593                const void  * writeme,
594                size_t        writemeLen,
595                int           isPieceData )
596{
597    struct tr_datatype * datatype;
598    assert( tr_amInEventThread( io->session ) );
599    dbgmsg( io, "adding %zu bytes into io->output", writemeLen );
600
601    datatype = tr_new( struct tr_datatype, 1 );
602    datatype->isPieceData = isPieceData != 0;
603    datatype->length = writemeLen;
604    tr_list_append( &io->output_datatypes, datatype );
605
606    evbuffer_add( tr_iobuf_output( io->iobuf ), writeme, writemeLen );
607    tr_iobuf_enable( io->iobuf, EV_WRITE );
608}
609
610void
611tr_peerIoWriteBuf( tr_peerIo         * io,
612                   struct evbuffer   * buf,
613                   int                 isPieceData )
614{
615    const size_t n = EVBUFFER_LENGTH( buf );
616
617    tr_peerIoWrite( io, EVBUFFER_DATA( buf ), n, isPieceData );
618    evbuffer_drain( buf, n );
619}
620
621/**
622***
623**/
624
625void
626tr_peerIoWriteBytes( tr_peerIo *       io,
627                     struct evbuffer * outbuf,
628                     const void *      bytes,
629                     size_t            byteCount )
630{
631    uint8_t * tmp;
632
633    switch( io->encryptionMode )
634    {
635        case PEER_ENCRYPTION_NONE:
636            evbuffer_add( outbuf, bytes, byteCount );
637            break;
638
639        case PEER_ENCRYPTION_RC4:
640            tmp = tr_new( uint8_t, byteCount );
641            tr_cryptoEncrypt( io->crypto, byteCount, bytes, tmp );
642            evbuffer_add( outbuf, tmp, byteCount );
643            tr_free( tmp );
644            break;
645
646        default:
647            assert( 0 );
648    }
649}
650
651void
652tr_peerIoWriteUint8( tr_peerIo *       io,
653                     struct evbuffer * outbuf,
654                     uint8_t           writeme )
655{
656    tr_peerIoWriteBytes( io, outbuf, &writeme, sizeof( uint8_t ) );
657}
658
659void
660tr_peerIoWriteUint16( tr_peerIo *       io,
661                      struct evbuffer * outbuf,
662                      uint16_t          writeme )
663{
664    uint16_t tmp = htons( writeme );
665
666    tr_peerIoWriteBytes( io, outbuf, &tmp, sizeof( uint16_t ) );
667}
668
669void
670tr_peerIoWriteUint32( tr_peerIo *       io,
671                      struct evbuffer * outbuf,
672                      uint32_t          writeme )
673{
674    uint32_t tmp = htonl( writeme );
675
676    tr_peerIoWriteBytes( io, outbuf, &tmp, sizeof( uint32_t ) );
677}
678
679/***
680****
681***/
682
683void
684tr_peerIoReadBytes( tr_peerIo *       io,
685                    struct evbuffer * inbuf,
686                    void *            bytes,
687                    size_t            byteCount )
688{
689    assert( EVBUFFER_LENGTH( inbuf ) >= byteCount );
690
691    switch( io->encryptionMode )
692    {
693        case PEER_ENCRYPTION_NONE:
694            evbuffer_remove( inbuf, bytes, byteCount );
695            break;
696
697        case PEER_ENCRYPTION_RC4:
698            evbuffer_remove( inbuf, bytes, byteCount );
699            tr_cryptoDecrypt( io->crypto, byteCount, bytes, bytes );
700            break;
701
702        default:
703            assert( 0 );
704    }
705}
706
707void
708tr_peerIoReadUint8( tr_peerIo *       io,
709                    struct evbuffer * inbuf,
710                    uint8_t *         setme )
711{
712    tr_peerIoReadBytes( io, inbuf, setme, sizeof( uint8_t ) );
713}
714
715void
716tr_peerIoReadUint16( tr_peerIo *       io,
717                     struct evbuffer * inbuf,
718                     uint16_t *        setme )
719{
720    uint16_t tmp;
721
722    tr_peerIoReadBytes( io, inbuf, &tmp, sizeof( uint16_t ) );
723    *setme = ntohs( tmp );
724}
725
726void
727tr_peerIoReadUint32( tr_peerIo *       io,
728                     struct evbuffer * inbuf,
729                     uint32_t *        setme )
730{
731    uint32_t tmp;
732
733    tr_peerIoReadBytes( io, inbuf, &tmp, sizeof( uint32_t ) );
734    *setme = ntohl( tmp );
735}
736
737void
738tr_peerIoDrain( tr_peerIo *       io,
739                struct evbuffer * inbuf,
740                size_t            byteCount )
741{
742    uint8_t * tmp = tr_new( uint8_t, byteCount );
743
744    tr_peerIoReadBytes( io, inbuf, tmp, byteCount );
745    tr_free( tmp );
746}
747
748int
749tr_peerIoGetAge( const tr_peerIo * io )
750{
751    return time( NULL ) - io->timeCreated;
752}
Note: See TracBrowser for help on using the repository browser.