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

Last change on this file since 7186 was 7186, checked in by charles, 14 years ago

(libT) #1532: patch from jhujhiti to remove (unused) fast extensions

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