peer.c

libbt-1.04

# include "config.h"

#if !WIN32
#   include <sys/types.h>
#   include <netinet/in.h>
#   include <arpa/inet.h>
#   include <netdb.h>
#   include <sys/socket.h>
#   if HAVE_UNISTD_H
#      include <unistd.h>
#   endif
#   if HAVE_FCNTL_H
#      include <fcntl.h>
#   endif
#endif
#include <sys/types.h>
#include <string.h>
#ifdef HAVE_STRINGS_H
#   include <strings.h>
#endif
#include <errno.h>
#include <time.h>
#include <poll.h>
#include <assert.h>

#include "bterror.h"
#include "btmessage.h"
#include "peer.h"
#include "stream.h"
#include "bitset.h"
#include "context.h"
#include "segmenter.h"

#if WIN32
#   define EINPROGRESS     WSAEINPROGRESS   /* Operation now in progress */
#   define close(s) closesocket(s)
#   define int32_t signed int
#endif
#define REQMAX 10
#define DOWNLOADS 4
#define REQUEST_SIZE 16384  /* Default request size */
char g_filebuffer[MAXREQUEST];	/* This should be moved to context or allocated in process_queue */

btPeerset* btPeerset_create( btPeerset *pset) {
    if (!pset) {
	pset = (btPeerset *)btmalloc(sizeof(btPeerset));
    }
    memset(pset, 0, sizeof(btPeerset));
    return pset;
}

/* prototypes */
int
send_have( btPeer *peer, int piece);
int 
send_cancel( btPeer *peer, int piece, int offs, int len);
int
send_choke( btPeer *peer, int choke) ;

int peer_connect_complete( btContext* ctx, struct btPeer *p) {
    int error;
    int	errlen;

    errlen = sizeof(error);
    if (getsockopt( p->ios.fd, SOL_SOCKET, SO_ERROR, &error, &errlen)) {
	p->ios.error = errno;
	bts_perror(p->ios.error, "getsockopt");
	return p->ios.error;
    }
    if (error != 0) {
	errno = error;
	p->ios.error = error;
	bts_perror(p->ios.error, "connect_complete");
	p->state = PEER_ERROR;
	return p->ios.error;
    }
    p->state = PEER_OUTGOING;
    printf("%d: completed connection %s\n", p->ios.fd, inet_ntoa(p->ip));
    return 0;    
}

int peer_connect_request( btContext* ctx, struct btPeer *p) {
    int sock;
    struct sockaddr_in addr;
    long flags;
    int err;

    if (p == NULL) return -1;
    if (!memcmp(ctx->myid, p->id, IDSIZE)) return -1;

    /* create the socket */
    sock = socket( PF_INET, SOCK_STREAM, 0);
    if (sock == -1) {
	bts_perror(errno, "socket");
	return -1;
    }

    /* change socket to non-blocking */
#if WIN32
    flags = ioctlsocket( sock, FIONBIO, (unsigned long *) 1);
    if (flags != 0) {
        bts_perror(errno, "ioctlsocket");
        return -1;
    }
#else
    flags = fcntl( sock, F_GETFL);
    if (flags < 0) {
        bts_perror(errno, "fcntl F_GETFL");
        return -1;
    }
    
    flags |= O_NONBLOCK;
    if ( fcntl( sock, F_SETFL, flags)) {
        bts_perror(errno, "fcntl F_SETFL");
        return -1;
    }
#endif


    /* connect the socket */
    addr.sin_family = AF_INET;
    addr.sin_addr = p->ip;
    addr.sin_port = htons( p->port);
    err = connect( sock, (void *)&addr, sizeof(struct sockaddr_in));
    if (err) {
#if WIN32
        errno=WSAGetLastError();
#endif
	if (errno != EINPROGRESS) {
	    bts_perror(errno, "connect");
	    close(sock);
	    return -1;
	}
    }

    /* create a kStream for the socket */
    kStream_create( &p->ios, sock);
    /* index the peer by socket */
    ctx->sockpeer[sock]=p;

    if (err) return 1;
    return 0;
}

btPeer *peer_add( btContext *ctx, unsigned download, char *id, struct in_addr *ip, int port) {
    btDownload *dl=ctx->downloads[download];
    btPeerset *pset = &dl->peerset;
    int idx, err;
    struct btPeer *p;

    DIE_UNLESS(download<ctx->downloadcount);

    /* initialize status info */
    if (ctx_addstatus( ctx, TMPLOC)) {
	/* over the connection limit */
	return NULL;
    }


    /* initialize the peer structure */
    p = btcalloc(1, sizeof(struct btPeer));
    p->download=download;
    kBitSet_create(&p->blocks, dl->fileset.npieces);
    memcpy(p->id, id, IDSIZE);
    p->ip=*ip;
    p->port = port;
    p->currentPiece = NULL;
    p->remote.choked = 1;
    p->local.choked = 1;
    
    /* connect to the peer */
    idx = pset->len++;
    err = peer_connect_request( ctx, p);
/*    printf("peer_add(..., %s, %d) = %d\n", ip, port, idx); */

    /* index the peer by peerid */
    pset->peer = btrealloc( pset->peer, sizeof(struct btPeer*) * pset->len);
    pset->peer[ idx] = p;

    /* fix the statmap link from fd to status */
    ctx_fixtmp( ctx, p->ios.fd);

    /* check connect results */
    if (err) {
	if (err == 1) {
	    /* incomplete */
	    printf("%d: %s:%d : incomplete\n", p->ios.fd, inet_ntoa(p->ip), p->port);
	    p->state=PEER_INIT;
	    ctx_setevents( ctx, p->ios.fd, POLLOUT);
	} else {
	    /* error connecting */
	    bts_perror(errno, "peer_connect failed");
	    p->state=PEER_ERROR;
	    p->local.unreachable=1;
	}
    } else {
	ctx_setevents( ctx, p->ios.fd, POLLIN);
	p->state=PEER_OUTGOING;
	if (peer_send_handshake( ctx, p)<0) {
	    p->state=PEER_ERROR;
	}
	if (peer_send_bitfield( ctx, p)<0) {
	    p->state=PEER_ERROR;
	}
    }

    return p;
}

int peer_send_handshake( btContext *ctx, btPeer *peer) {
    /*
    *    Handshake:
    *    \x13BitTorrent protocol\0\0\0\0\0\0\0\0<sha1 info hash><20byte peerid>
    */
    char shake[0x14] = "\x13" "BitTorrent protocol";
    char flags[8] = { 0 };
    int err;
#if 0
    printf("peer_send_handshake\n");
#endif
    assert(peer->download >= 0);
    err = kStream_fwrite( &peer->ios, shake, 0x14);
    if (err < 0) return -1;
    err = kStream_fwrite( &peer->ios, flags, 8);
    if (err < 0) return -2;
    err = kStream_fwrite( &peer->ios, ctx->downloads[peer->download]->infohash, SHA_DIGEST_LENGTH);
    if (err < 0) return -3;
    err = kStream_fwrite( &peer->ios, ctx->myid, IDSIZE);
    if (err < 0) return -4;
    return 0;
}

/*
 * check_handshake()
 *
 * Returns:
 *     0 - ok or incomplete
 *    -1 - error
 * Sets ios.error to:
 *    BTERR_PROTOCOL_ID if there is an error in the protocol identification
 *    BTERR_UNKNOWN_FLAGS if there the flags are set incorrectly
 *    BTERR_HASH_MISMATCH if the infohash doesn't match ours
 *
 * Sets peer->id to:
 *    remote id
 */
#define PROTO_LENGTH 0x14
#define FLAGS_LENGTH 0x08
static int check_handshake( btContext *ctx, btPeer *peer, char *shake, int len) 
{
    char *flags = shake + PROTO_LENGTH;
    char *infohash = flags + FLAGS_LENGTH;
    char *id = infohash + SHA_DIGEST_LENGTH;
    if (shake[0] != '\x13') {
	peer->ios.error = BTERR_PROTOCOL_ID;
	return -1;
    }
    if (len >= PROTO_LENGTH 
	    && memcmp( shake, "\x13" "BitTorrent protocol", PROTO_LENGTH) != 0) 
    {
	/* bad protocol */
	peer->ios.error = BTERR_PROTOCOL_ID;
	return -1;
    }
    if (len >= PROTO_LENGTH + FLAGS_LENGTH 
	    && memcmp( flags, "\0\0\0\0\0\0\0\0", FLAGS_LENGTH)!= 0) 
    {
	/* bad flags */
	int i;
	printf("Unknown flags in handshake: ");
	for (i=0; i<8; i++) {
	    printf("%02x", flags[i]);
	}
	printf("\n");
    }
    if (len >= PROTO_LENGTH + FLAGS_LENGTH + SHA_DIGEST_LENGTH)
    {
        int i=INT_MAX;

	if(peer->state==PEER_INCOMING && peer->download==INT_MAX)
	{	/* Find the correct torrent */
	    for(i=0; i<ctx->downloadcount; i++)
	    {
		if (memcmp( infohash, ctx->downloads[i]->infohash, SHA_DIGEST_LENGTH)==0) 
		{
		    int idx;
		    btPeerset *pset = &ctx->downloads[i]->peerset;

		    /* index the peer by peerid */
		    idx = pset->len++;
		    pset->peer = btrealloc( pset->peer, sizeof(struct btPeer*) * pset->len);
		    pset->peer[ idx] = peer;

		    peer->download=i;

		    kBitSet_create(&peer->blocks, ctx->downloads[i]->fileset.npieces);
		    if (peer_send_handshake( ctx, peer)<0 || peer_send_bitfield( ctx, peer)<0)
		    {
		        peer->state=PEER_ERROR;
			return -1;
		    }
		    break;
		}
	    }
	}
	else
	{
	    i=peer->download;
	    if (memcmp( infohash, ctx->downloads[i]->infohash, SHA_DIGEST_LENGTH)!=0)
	        i=INT_MAX;
	}
	if(i>=ctx->downloadcount)
	{
	    /* bad infohash from the peer */
	    peer->ios.error = BTERR_HASH_MISMATCH;
	    return -1;
	}
    }
    if (len >= PROTO_LENGTH + FLAGS_LENGTH + SHA_DIGEST_LENGTH + IDSIZE) {
	memcpy( peer->id, id, IDSIZE);
    }
    return 0;
}

/* 
 * recv_handshake()
 *
 * Returns 
 *   0 success
 *  -1 error
 *
 * peer->ios.error iff error return is set to one of
 *   EAGAIN - retry later
 *   some error code
 *
 * peer->state is set to one of
 *   PEER_ERROR - handshake failed
 *   PEER_CONNECT - handshake is still pending
 *   PEER_GOOD - handshake successful
 */
int recv_handshake( btContext *ctx, btPeer *peer) {
    char shake[PROTO_LENGTH + FLAGS_LENGTH + SHA_DIGEST_LENGTH + IDSIZE];
    int err;
    err = kStream_fread( &peer->ios, shake, sizeof(shake));
    if (err < 0) {
	if (peer->ios.error == EAGAIN) {
	    /* can't get whole handshake; try for partial handshake */
	    int len = kStream_iqlen(&peer->ios);
	    if (len > 0) {
		err = kStream_fpeek( &peer->ios, shake, len);
		DIE_UNLESS(err == len);
		if (check_handshake( ctx, peer, shake, len)) {
		    peer->state = PEER_ERROR;
		    return -1;
		}
	    }
	    peer->ios.error = EAGAIN;
	}
	return -1;
    }
    DIE_UNLESS(err == sizeof(shake));
    if (check_handshake( ctx, peer, shake, sizeof(shake))) {
	return -1;
    }
    peer->state = PEER_GOOD;
    ctx->downloads[peer->download]->peerset.incomplete++;
#if 0
    printf("%d: got handshake\n", peer->ios.fd);
#endif
    return 0;
}

/*
 * queue_request()
 */
static int
queue_request( btRequestQueue *q, int piece, int offset, int len) {
    int oldtail = q->tail;
    int newtail = (oldtail+1) % QUEUESIZE;
    if (len > MAXREQUEST) return -1;
    if (newtail == q->head) return -2;
    q->req[oldtail].block = piece;
    q->req[oldtail].offset = offset;
    q->req[oldtail].length = len;
    q->tail = newtail;
    return 0;
}

static int 
clear_request_queue( btRequestQueue *q) {
    q->head = q->tail;
    return 0;
}

/* returns number of queued requests that were deleted (usually 1) */
static int 
remove_queued_request( btRequestQueue *q, int block, int offset, int len) {
    btRequest *r;
    int i = q->head;
    int offby=0;
    while (i != q->tail) {
	r = &q->req[i];

        if (r->block == block &&
	    r->offset == offset &&
	    r->length == len) {
	    offby++;
	}
	if (offby) {
            q->req[i] = q->req[(i+offby)%QUEUESIZE];
	}
	i = (i+1) % QUEUESIZE;
    }
    q->tail = (q->tail + QUEUESIZE - offby) % QUEUESIZE;
    return offby;
}

static btRequest *dequeue_request( btRequestQueue *q) {
    btRequest *req;
    int oldhead = q->head;
    int newhead = (oldhead+1) % QUEUESIZE;
    if (oldhead == q->tail) return NULL;
    req = &q->req[oldhead];
    q->head = newhead;
    return req;
}

static int
queue_len( btRequestQueue *q ) {
    int ct=q->tail - q->head;
    if (ct < 0) ct += QUEUESIZE;
    return ct;
}

static void start_rate_timer( btPeerStatus *ps, time_t now) {
    if (ps->send_time == 0) {
	ps->send_time = now;
    }
}

static void stop_rate_timer( btPeerStatus *ps, time_t now) {
    if (ps->send_time != 0) {
	ps->total_time += now - ps->send_time;
	ps->send_time = 0;
    }
}

static int rate_timer( btPeerStatus *ps, time_t now) {
    int total;
    total = ps->total_time;
    if (ps->send_time != 0) {
	total += now - ps->send_time;
    }
    if (total < 1) total = 1;
    return total;
}

#define SHIFT_INT32(ptr,nbo,ival) \
    (nbo=htonl(ival), memcpy(ptr,&nbo,sizeof(int32_t)), ptr+=sizeof(int32_t))
#define SHIFT_BYTE(ptr,ival) ((*((unsigned char *)(ptr))++) = ival)

#define UNSHIFT_INT32(ptr,nbo,ival) \
    (memcpy(&nbo,ptr,sizeof(int32_t)), ival=ntohl(nbo), ptr+=sizeof(int32_t))
#define UNSHIFT_BYTE(ptr,ival) (ival = (*((unsigned char *)(ptr))++)
/*
 * Return 1 if there are more messages waiting
 * Return 0 on success, 
 * return -1 on error  (peer->ios.error contains the error code)
 * return -2 on unknown message.
 */
int
recv_peermsg( btContext *ctx, btPeer *peer) {
    btDownload *dl = ctx->downloads[peer->download];
    int32_t nbo_len;
    int len;
    char msg[80];
    char *nmsg, *param;
    int res = 0;
    int err;
    int32_t nbo;

    DIE_UNLESS(peer->download<ctx->downloadcount);

    err = kStream_fpeek( &peer->ios, (char *)&nbo_len, sizeof(nbo_len));