X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/eeec4714ded8d0a6be32567502577faaca925a18..1059885582cc8001c4dbc242aa94a1ed1ae5e2c8:/examples/msg/app-bittorrent/peer.c diff --git a/examples/msg/app-bittorrent/peer.c b/examples/msg/app-bittorrent/peer.c index 4aaf7c018f..427df2b7f5 100644 --- a/examples/msg/app-bittorrent/peer.c +++ b/examples/msg/app-bittorrent/peer.c @@ -1,5 +1,4 @@ -/* Copyright (c) 2012-2016. The SimGrid Team. - * All rights reserved. */ +/* Copyright (c) 2012-2017. The SimGrid Team. All rights reserved. */ /* This program is free software; you can redistribute it and/or modify it * under the terms of the license (GNU LGPL) which comes with this package. */ @@ -14,65 +13,85 @@ XBT_LOG_NEW_DEFAULT_CATEGORY(msg_peers, "Messages specific for the peers"); -//TODO: Let users change this /* - * File transfered data - * For the test, default values are : + * User parameters for transferred file data. For the test, the default values are : * File size: 10 pieces * 5 blocks/piece * 16384 bytes/block = 819200 bytes */ - -#define FILE_PIECES 10 -#define PIECES_BLOCKS 5 +#define FILE_PIECES 10U +#define PIECES_BLOCKS 5U #define BLOCK_SIZE 16384 -#define ENABLE_END_GAME_MODE 1 +static const unsigned long int FILE_SIZE = FILE_PIECES * PIECES_BLOCKS * BLOCK_SIZE; /** Number of blocks asked by each request */ #define BLOCKS_REQUESTED 2 -static const unsigned long int FILE_SIZE = FILE_PIECES * PIECES_BLOCKS * BLOCK_SIZE; - +#define ENABLE_END_GAME_MODE 1 #define SLEEP_DURATION 1 +int count_pieces(unsigned int bitfield){ + int count=0; + unsigned int n = bitfield; + while (n){ + count += n & 1U; + n >>= 1U ; + } + return count; +} + +int peer_has_not_piece(peer_t peer, unsigned int piece){ + return !(peer->bitfield & 1U<current_pieces & 1U<=0; i--) + (*status)[i] = (bitfield & (1U << i)) ? '1' : '0'; + (*status)[FILE_PIECES] = '\0'; +} + /** Peer main function */ int peer(int argc, char *argv[]) { - s_peer_t peer; //Check arguments xbt_assert(argc == 3 || argc == 4, "Wrong number of arguments"); + //Build peer object - if (argc == 4) { - peer_init(&peer, xbt_str_parse_int(argv[1],"Invalid ID: %s"), 1); - } else { - peer_init(&peer, xbt_str_parse_int(argv[1],"Invalid ID: %s"), 0); - } + peer_t peer = peer_init(xbt_str_parse_int(argv[1],"Invalid ID: %s"), argc==4 ? 1:0); + //Retrieve deadline double deadline = xbt_str_parse_double(argv[2],"Invalid deadline: %s"); xbt_assert(deadline > 0, "Wrong deadline supplied"); - XBT_INFO("Hi, I'm joining the network with id %d", peer.id); + + char *status = xbt_malloc0(FILE_PIECES+1); + get_status(&status, peer->bitfield); + XBT_INFO("Hi, I'm joining the network with id %d", peer->id); //Getting peer data from the tracker. - if (get_peers_data(&peer)) { - XBT_DEBUG("Got %d peers from the tracker", xbt_dict_length(peer.peers)); - XBT_DEBUG("Here is my current status: %s", peer.bitfield); - peer.begin_receive_time = MSG_get_clock(); - MSG_mailbox_set_async(peer.mailbox); - if (has_finished(peer.bitfield)) { - peer.pieces = FILE_PIECES; - send_handshake_all(&peer); - seed_loop(&peer, deadline); + if (get_peers_data(peer)) { + XBT_DEBUG("Got %d peers from the tracker. Current status is: %s", xbt_dict_length(peer->peers), status); + peer->begin_receive_time = MSG_get_clock(); + MSG_mailbox_set_async(peer->mailbox); + if (has_finished(peer->bitfield)) { + send_handshake_all(peer); } else { - leech_loop(&peer, deadline); - seed_loop(&peer, deadline); + leech_loop(peer, deadline); } + seed_loop(peer, deadline); } else { XBT_INFO("Couldn't contact the tracker."); } - XBT_INFO("Here is my current status: %s", peer.bitfield); - if (peer.comm_received) { - MSG_comm_destroy(peer.comm_received); + get_status(&status, peer->bitfield); + XBT_INFO("Here is my current status: %s", status); + if (peer->comm_received) { + MSG_comm_destroy(peer->comm_received); } - peer_free(&peer); + xbt_free(status); + peer_free(peer); return 0; } @@ -84,14 +103,12 @@ void leech_loop(peer_t peer, double deadline) { double next_choked_update = MSG_get_clock() + UPDATE_CHOKED_INTERVAL; XBT_DEBUG("Start downloading."); - /* - * Send a "handshake" message to all the peers it got - * (since it couldn't have gotten more than 50 peers) - */ + + /* Send a "handshake" message to all the peers it got (since it couldn't have gotten more than 50 peers) */ send_handshake_all(peer); XBT_DEBUG("Starting main leech loop"); - while (MSG_get_clock() < deadline && peer->pieces < FILE_PIECES) { + while (MSG_get_clock() < deadline && count_pieces(peer->bitfield) < FILE_PIECES) { if (peer->comm_received == NULL) { peer->task_received = NULL; peer->comm_received = MSG_task_irecv(&peer->task_received, peer->mailbox); @@ -105,7 +122,7 @@ void leech_loop(peer_t peer, double deadline) } } else { //We don't execute the choke algorithm if we don't already have a piece - if (MSG_get_clock() >= next_choked_update && peer->pieces > 0) { + if (MSG_get_clock() >= next_choked_update && count_pieces(peer->bitfield) > 0) { update_choked_peers(peer); next_choked_update += UPDATE_CHOKED_INTERVAL; } else { @@ -113,9 +130,8 @@ void leech_loop(peer_t peer, double deadline) } } } - if (peer->pieces == FILE_PIECES) + if (has_finished(peer->bitfield)) XBT_DEBUG("%d becomes a seeder", peer->id); - } /** @brief Peer main loop when it is seeding @@ -151,30 +167,30 @@ void seed_loop(peer_t peer, double deadline) } } -/** @brief Retrieves the peer list from the tracker +/** @brief Retrieves the peer list from the tracker * @param peer current peer data */ int get_peers_data(peer_t peer) { int success = 0; - int send_success = 0; double timeout = MSG_get_clock() + GET_PEERS_TIMEOUT; + //Build the task to send to the tracker tracker_task_data_t data = tracker_task_data_new(MSG_host_get_name(MSG_host_self()), peer->mailbox_tracker, peer->id, 0, 0, FILE_SIZE); - //Build the task to send. msg_task_t task_send = MSG_task_create(NULL, 0, TRACKER_COMM_SIZE, data); - msg_task_t task_received = NULL; - msg_comm_t comm_received; - while ((send_success == 0) && MSG_get_clock() < timeout) { + while ((success == 0) && MSG_get_clock() < timeout) { XBT_DEBUG("Sending a peer request to the tracker."); msg_error_t status = MSG_task_send_with_timeout(task_send, TRACKER_MAILBOX, GET_PEERS_TIMEOUT); if (status == MSG_OK) { - send_success = 1; + success = 1; } } - while ((success ==0) && MSG_get_clock() < timeout) { - comm_received = MSG_task_irecv(&task_received, peer->mailbox_tracker); + + success = 0; + msg_task_t task_received = NULL; + while ((success == 0) && MSG_get_clock() < timeout) { + msg_comm_t comm_received = MSG_task_irecv(&task_received, peer->mailbox_tracker); msg_error_t status = MSG_comm_wait(comm_received, GET_PEERS_TIMEOUT); if (status == MSG_OK) { tracker_task_data_t data = MSG_task_get_data(task_received); @@ -190,7 +206,6 @@ int get_peers_data(peer_t peer) MSG_comm_destroy(comm_received); tracker_task_data_free(data); MSG_task_destroy(task_received); - comm_received = NULL; } } @@ -202,36 +217,35 @@ int get_peers_data(peer_t peer) * @param id id of the peer to take in the network * @param seed indicates if the peer is a seed. */ -void peer_init(peer_t peer, int id, int seed) +peer_t peer_init(int id, int seed) { + peer_t peer = xbt_new(s_peer_t,1); peer->id = id; + peer->hostname = MSG_host_get_name(MSG_host_self()); + snprintf(peer->mailbox,MAILBOX_SIZE-1, "%d", id); snprintf(peer->mailbox_tracker,MAILBOX_SIZE-1, "tracker_%d", id); - peer->peers = xbt_dict_new(); - peer->active_peers = xbt_dict_new(); - peer->hostname = MSG_host_get_name(MSG_host_self()); + peer->peers = xbt_dict_new_homogeneous(NULL); + peer->active_peers = xbt_dict_new_homogeneous(NULL); - peer->bitfield = xbt_new(char, FILE_PIECES + 1); - peer->bitfield_blocks = xbt_new(char, (FILE_PIECES) * (PIECES_BLOCKS) + 1); if (seed) { - memset(peer->bitfield, '1', sizeof(char) * (FILE_PIECES + 1)); - memset(peer->bitfield_blocks, '1', sizeof(char) * FILE_PIECES * (PIECES_BLOCKS)); + peer->bitfield = (1U<bitfield_blocks = (1ULL<<(FILE_PIECES * PIECES_BLOCKS))-1ULL ; } else { - memset(peer->bitfield, '0', sizeof(char) * (FILE_PIECES + 1)); - memset(peer->bitfield_blocks, '0', sizeof(char) * FILE_PIECES * (PIECES_BLOCKS)); + peer->bitfield = 0; + peer->bitfield_blocks = 0; } - peer->bitfield[FILE_PIECES] = '\0'; - peer->pieces = 0; + peer->current_pieces = 0; peer->pieces_count = xbt_new0(short, FILE_PIECES); - peer->current_pieces = xbt_dynar_new(sizeof(int), NULL); - - peer->stream = (RngStream)MSG_host_get_property_value(MSG_host_self(), "stream"); + peer->stream = (RngStream)MSG_host_get_data(MSG_host_self()); peer->comm_received = NULL; peer->round = 0; + + return peer; } /** Destroys a poor peer object. */ @@ -245,18 +259,16 @@ void peer_free(peer_t peer) } xbt_dict_free(&peer->peers); xbt_dict_free(&peer->active_peers); - xbt_dynar_free(&peer->current_pieces); xbt_free(peer->pieces_count); - xbt_free(peer->bitfield); - xbt_free(peer->bitfield_blocks); + xbt_free(peer); } -/** @brief Returns if a peer has finished downloading the file +/** @brief Returns if a peer has finished downloading the file * @param bitfield peer bitfield */ -int has_finished(char *bitfield) +int has_finished(unsigned int bitfield) { - return ((memchr(bitfield, '0', sizeof(char) * FILE_PIECES) == NULL) ? 1 : 0); + return bitfield == (1U<type], message->mailbox, + message->issuer_host_name); + connection_t remote_peer; remote_peer = xbt_dict_get_or_null_ext(peer->peers, (char *) &message->peer_id, sizeof(int)); + switch (message->type) { case MESSAGE_HANDSHAKE: - XBT_DEBUG("Received a HANDSHAKE from %s (%s)", message->mailbox, message->issuer_host_name); //Check if the peer is in our connection list. if (remote_peer == 0) { xbt_dict_set_ext(peer->peers, (char *) &message->peer_id, sizeof(int), connection_new(message->peer_id), NULL); @@ -303,11 +320,10 @@ void handle_message(peer_t peer, msg_task_t task) send_bitfield(peer, message->mailbox); break; case MESSAGE_BITFIELD: - XBT_DEBUG("Recieved a BITFIELD message from %s (%s)", message->mailbox, message->issuer_host_name); //Update the pieces list update_pieces_count_from_bitfield(peer, message->bitfield); //Store the bitfield - remote_peer->bitfield = xbt_strdup(message->bitfield); + remote_peer->bitfield = message->bitfield; xbt_assert(!remote_peer->am_interested, "Should not be interested at first"); if (is_interested(peer, remote_peer)) { remote_peer->am_interested = 1; @@ -315,22 +331,19 @@ void handle_message(peer_t peer, msg_task_t task) } break; case MESSAGE_INTERESTED: - XBT_DEBUG("Recieved an INTERESTED message from %s (%s)", message->mailbox, message->issuer_host_name); xbt_assert((remote_peer != NULL), "A non-in-our-list peer has sent us a message. WTH ?"); //Update the interested state of the peer. remote_peer->interested = 1; update_active_peers_set(peer, remote_peer); break; case MESSAGE_NOTINTERESTED: - XBT_DEBUG("Received a NOTINTERESTED message from %s (%s)", message->mailbox, message->issuer_host_name); xbt_assert((remote_peer != NULL), "A non-in-our-list peer has sent us a message. WTH ?"); remote_peer->interested = 0; update_active_peers_set(peer, remote_peer); break; case MESSAGE_UNCHOKE: xbt_assert((remote_peer != NULL), "A non-in-our-list peer has sent us a message. WTH ?"); - XBT_DEBUG("Received a UNCHOKE message from %s (%s)", message->mailbox, message->issuer_host_name); - xbt_assert(remote_peer->choked_download, "WTF !!!"); + xbt_assert(remote_peer->choked_download); remote_peer->choked_download = 0; //Send requests to the peer, since it has unchoked us if (remote_peer->am_interested) @@ -338,21 +351,18 @@ void handle_message(peer_t peer, msg_task_t task) break; case MESSAGE_CHOKE: xbt_assert((remote_peer != NULL), "A non-in-our-list peer has sent us a message. WTH ?"); - XBT_DEBUG("Received a CHOKE message from %s (%s)", message->mailbox, message->issuer_host_name); - xbt_assert(!remote_peer->choked_download, "WTF !!!"); + xbt_assert(!remote_peer->choked_download); remote_peer->choked_download = 1; - remove_current_piece(peer, remote_peer, remote_peer->current_piece); + if (remote_peer->current_piece != -1) + remove_current_piece(peer, remote_peer, remote_peer->current_piece); break; case MESSAGE_HAVE: - XBT_DEBUG("Received a HAVE message from %s (%s) of piece %d", message->mailbox, message->issuer_host_name, - message->index); - xbt_assert(remote_peer->bitfield, "bitfield not received"); + XBT_DEBUG("\t for piece %d", message->index); xbt_assert((message->index >= 0 && message->index < FILE_PIECES), "Wrong HAVE message received"); - remote_peer->bitfield[message->index] = '1'; + remote_peer->bitfield = remote_peer->bitfield | (1U<index); peer->pieces_count[message->index]++; //If the piece is in our pieces, we tell the peer that we are interested. - if ((remote_peer->am_interested == 0) && - peer->bitfield[message->index] == '0') { + if ((remote_peer->am_interested == 0) && peer_has_not_piece(peer,message->index)) { remote_peer->am_interested = 1; send_interested(peer, message->mailbox); if (remote_peer->choked_download == 0) @@ -360,54 +370,53 @@ void handle_message(peer_t peer, msg_task_t task) } break; case MESSAGE_REQUEST: - xbt_assert(remote_peer->interested, "WTF !!!"); - + xbt_assert(remote_peer->interested); xbt_assert((message->index >= 0 && message->index < FILE_PIECES), "Wrong request received"); if (remote_peer->choked_upload == 0) { - XBT_DEBUG("Received a REQUEST from %s (%s) for %d (%d,%d)", message->mailbox, message->issuer_host_name, - message->index, message->block_index, message->block_index + message->block_length); - if (peer->bitfield[message->index] == '1') { + XBT_DEBUG("\t for piece %d (%d,%d)", message->index, message->block_index, + message->block_index + message->block_length); + if (!peer_has_not_piece(peer, message->index)) { send_piece(peer, message->mailbox, message->index, message->block_index, message->block_length); } } else { - XBT_DEBUG("Received a REQUEST from %s (%s) for %d but he is choked.", message->mailbox, - message->issuer_host_name, message->peer_id); + XBT_DEBUG("\t for piece %d but he is choked.", message->peer_id); } break; case MESSAGE_PIECE: - XBT_DEBUG("Received piece %d (%d,%d) from %s (%s)", message->index, message->block_index, - message->block_index + message->block_length, message->mailbox, message->issuer_host_name); - xbt_assert(!remote_peer->choked_download, "WTF !!!"); + XBT_DEBUG(" \t for piece %d (%d,%d)", message->index, message->block_index, + message->block_index + message->block_length); + xbt_assert(!remote_peer->choked_download); xbt_assert(remote_peer->am_interested || ENABLE_END_GAME_MODE, "Can't received a piece if I'm not interested wihtout end-game mode!" - "piece (%d) bitfield(%s) remote bitfield(%s)", message->index, peer->bitfield, remote_peer->bitfield); + "piece (%d) bitfield(%u) remote bitfield(%u)", message->index, peer->bitfield, remote_peer->bitfield); xbt_assert(remote_peer->choked_download != 1, "Can't received a piece if I'm choked !"); xbt_assert((message->index >= 0 && message->index < FILE_PIECES), "Wrong piece received"); //TODO: Execute à computation. - if (peer->bitfield[message->index] == '0') { - update_bitfield_blocks(peer, message->index, message->block_index, message->block_length); - if (piece_complete(peer, message->index)) { - //Removing the piece from our piece list - remove_current_piece(peer, remote_peer, message->index); - //Setting the fact that we have the piece - peer->bitfield[message->index] = '1'; - peer->pieces++; - XBT_DEBUG("My status is now %s", peer->bitfield); - //Sending the information to all the peers we are connected to - send_have(peer, message->index); - //sending UNINTERSTED to peers that doesn't have what we want. - update_interested_after_receive(peer); - } else { // piece not completed - send_request_to_peer(peer, remote_peer, message->index); // ask for the next block - } - } else { - XBT_DEBUG("However, we already have it"); - xbt_assert(ENABLE_END_GAME_MODE, "Should not happen because we don't use end game mode !"); - request_new_piece_to_peer(peer, remote_peer); + if (peer_has_not_piece(peer,message->index)) { + update_bitfield_blocks(peer, message->index, message->block_index, message->block_length); + if (piece_complete(peer, message->index)) { + //Removing the piece from our piece list + remove_current_piece(peer, remote_peer, message->index); + //Setting the fact that we have the piece + peer->bitfield = peer->bitfield | (1U<index); + char* status = xbt_malloc0(FILE_PIECES+1); + get_status(&status, peer->bitfield); + XBT_DEBUG("My status is now %s", status); + xbt_free(status); + //Sending the information to all the peers we are connected to + send_have(peer, message->index); + //sending UNINTERESTED to peers that do not have what we want. + update_interested_after_receive(peer); + } else { // piece not completed + send_request_to_peer(peer, remote_peer, message->index); // ask for the next block } + } else { + XBT_DEBUG("However, we already have it"); + xbt_assert(ENABLE_END_GAME_MODE, "Should not happen because we don't use end game mode !"); + request_new_piece_to_peer(peer, remote_peer); + } break; case MESSAGE_CANCEL: - XBT_DEBUG("The received CANCEL from %s (%s)", message->mailbox, message->issuer_host_name); break; default: THROW_IMPOSSIBLE; @@ -426,17 +435,15 @@ void request_new_piece_to_peer(peer_t peer, connection_t remote_peer) { int piece = select_piece_to_download(peer, remote_peer); if (piece != -1) { - xbt_dynar_push_as(peer->current_pieces, int, piece); + peer->current_pieces|= (1U << (unsigned int) piece); send_request_to_peer(peer, remote_peer, piece); } } /** remove current_piece from the list of currently downloaded pieces. */ -void remove_current_piece(peer_t peer, connection_t remote_peer, int current_piece) +void remove_current_piece(peer_t peer, connection_t remote_peer, unsigned int current_piece) { - int piece_index = xbt_dynar_search_or_negative(peer->current_pieces, ¤t_piece); - if (piece_index != -1) - xbt_dynar_remove_at(peer->current_pieces, piece_index, NULL); + peer->current_pieces &= ~(1U << current_piece); remote_peer->current_piece = -1; } @@ -444,11 +451,10 @@ void remove_current_piece(peer_t peer, connection_t remote_peer, int current_pie * @param peer peer we want to update the list * @param bitfield bitfield */ -void update_pieces_count_from_bitfield(peer_t peer, char *bitfield) +void update_pieces_count_from_bitfield(peer_t peer, unsigned int bitfield) { - int i; - for (i = 0; i < FILE_PIECES; i++) { - if (bitfield[i] == '1') { + for (int i = 0; i < FILE_PIECES; i++) { + if (bitfield & (1U << i)) { peer->pieces_count[i]++; } } @@ -466,32 +472,30 @@ void update_pieces_count_from_bitfield(peer_t peer, char *bitfield) */ int select_piece_to_download(peer_t peer, connection_t remote_peer) { - int piece = -1; - - piece = partially_downloaded_piece(peer, remote_peer); + int piece = partially_downloaded_piece(peer, remote_peer); // strict priority policy if (piece != -1) return piece; // end game mode - if (xbt_dynar_length(peer->current_pieces) >= (FILE_PIECES - peer->pieces) && + if (count_pieces(peer->current_pieces) >= (FILE_PIECES - count_pieces(peer->bitfield)) && (is_interested(peer, remote_peer) != 0)) { - if(ENABLE_END_GAME_MODE == 0) +#if ENABLE_END_GAME_MODE == 0 return -1; - int i; +#endif int nb_interesting_pieces = 0; - int random_piece_index, current_index = 0; // compute the number of interesting pieces - for (i = 0; i < FILE_PIECES; i++) { - if (peer->bitfield[i] == '0' && remote_peer->bitfield[i] == '1') { + for (int i = 0; i < FILE_PIECES; i++) { + if (peer_has_not_piece(peer,i) && connection_has_piece(remote_peer,i)) { nb_interesting_pieces++; } } - xbt_assert(nb_interesting_pieces != 0, "WTF !!!"); + xbt_assert(nb_interesting_pieces != 0); // get a random interesting piece - random_piece_index = RngStream_RandInt(peer->stream, 0, nb_interesting_pieces - 1); - for (i = 0; i < FILE_PIECES; i++) { - if (peer->bitfield[i] == '0' && remote_peer->bitfield[i] == '1') { + int random_piece_index = RngStream_RandInt(peer->stream, 0, nb_interesting_pieces - 1); + int current_index = 0; + for (int i = 0; i < FILE_PIECES; i++) { + if (peer_has_not_piece(peer,i) && connection_has_piece(remote_peer,i)) { if (random_piece_index == current_index) { piece = i; break; @@ -499,27 +503,24 @@ int select_piece_to_download(peer_t peer, connection_t remote_peer) current_index++; } } - xbt_assert(piece != -1, "WTF !!!"); + xbt_assert(piece != -1); return piece; } // Random first policy - if (peer->pieces < 4 && (is_interested_and_free(peer, remote_peer) != 0)) { - int i; + if (count_pieces(peer->bitfield) < 4 && (is_interested_and_free(peer, remote_peer) != 0)) { int nb_interesting_pieces = 0; - int random_piece_index, current_index = 0; // compute the number of interesting pieces - for (i = 0; i < FILE_PIECES; i++) { - if (peer->bitfield[i] == '0' && remote_peer->bitfield[i] == '1' && - (in_current_pieces(peer, i) == 0)) { + for (int i = 0; i < FILE_PIECES; i++) { + if (peer_has_not_piece(peer, i) && connection_has_piece(remote_peer,i) && peer_is_not_downloading_piece(peer, i)) { nb_interesting_pieces++; } } - xbt_assert(nb_interesting_pieces != 0, "WTF !!!"); + xbt_assert(nb_interesting_pieces != 0); // get a random interesting piece - random_piece_index = RngStream_RandInt(peer->stream, 0, nb_interesting_pieces - 1); - for (i = 0; i < FILE_PIECES; i++) { - if (peer->bitfield[i] == '0' && remote_peer->bitfield[i] == '1' && - (in_current_pieces(peer, i) == 0)) { + int random_piece_index = RngStream_RandInt(peer->stream, 0, nb_interesting_pieces - 1); + int current_index = 0; + for (int i = 0; i < FILE_PIECES; i++) { + if (peer_has_not_piece(peer, i) && connection_has_piece(remote_peer,i) && peer_is_not_downloading_piece(peer, i)) { if (random_piece_index == current_index) { piece = i; break; @@ -527,34 +528,31 @@ int select_piece_to_download(peer_t peer, connection_t remote_peer) current_index++; } } - xbt_assert(piece != -1, "WTF !!!"); + xbt_assert(piece != -1); return piece; } else { // Rarest first policy - int i; short min = SHRT_MAX; int nb_min_pieces = 0; - int random_rarest_index, current_index = 0; + int current_index = 0; // compute the smallest number of copies of available pieces - for (i = 0; i < FILE_PIECES; i++) { - if (peer->pieces_count[i] < min && peer->bitfield[i] == '0' && - remote_peer->bitfield[i] == '1' && (in_current_pieces(peer, i) == 0)) + for (int i = 0; i < FILE_PIECES; i++) { + if (peer->pieces_count[i] < min && peer_has_not_piece(peer, i) && connection_has_piece(remote_peer,i) && + peer_is_not_downloading_piece(peer, i)) min = peer->pieces_count[i]; } - xbt_assert(min != SHRT_MAX || - (is_interested_and_free(peer, remote_peer) ==0), "WTF !!!"); + xbt_assert(min != SHRT_MAX || (is_interested_and_free(peer, remote_peer) ==0)); // compute the number of rarest pieces - for (i = 0; i < FILE_PIECES; i++) { - if (peer->pieces_count[i] == min && peer->bitfield[i] == '0' && - remote_peer->bitfield[i] == '1' && (in_current_pieces(peer, i) ==0)) + for (int i = 0; i < FILE_PIECES; i++) { + if (peer->pieces_count[i] == min && peer_has_not_piece(peer, i) && connection_has_piece(remote_peer,i) && + peer_is_not_downloading_piece(peer, i)) nb_min_pieces++; } - xbt_assert(nb_min_pieces != 0 || - (is_interested_and_free(peer, remote_peer)==0), "WTF !!!"); + xbt_assert(nb_min_pieces != 0 || (is_interested_and_free(peer, remote_peer)==0)); // get a random rarest piece - random_rarest_index = RngStream_RandInt(peer->stream, 0, nb_min_pieces - 1); - for (i = 0; i < FILE_PIECES; i++) { - if (peer->pieces_count[i] == min && peer->bitfield[i] == '0' && - remote_peer->bitfield[i] == '1' && (in_current_pieces(peer, i)==0)) { + int random_rarest_index = RngStream_RandInt(peer->stream, 0, nb_min_pieces - 1); + for (int i = 0; i < FILE_PIECES; i++) { + if (peer->pieces_count[i] == min && peer_has_not_piece(peer, i) && connection_has_piece(remote_peer,i) && + peer_is_not_downloading_piece(peer, i)) { if (random_rarest_index == current_index) { piece = i; break; @@ -562,8 +560,7 @@ int select_piece_to_download(peer_t peer, connection_t remote_peer) current_index++; } } - xbt_assert(piece != -1 || - (is_interested_and_free(peer, remote_peer) == 0), "WTF !!!"); + xbt_assert(piece != -1 || (is_interested_and_free(peer, remote_peer) == 0)); return piece; } } @@ -578,7 +575,8 @@ void update_choked_peers(peer_t peer) XBT_DEBUG("(%d) update_choked peers %d active peers", peer->id, xbt_dict_size(peer->active_peers)); //update the current round peer->round = (peer->round + 1) % 3; - char *key, *key_choked=NULL; + char *key; + char *key_choked=NULL; connection_t peer_choosed = NULL; connection_t peer_choked = NULL; //remove a peer from the list @@ -591,7 +589,7 @@ void update_choked_peers(peer_t peer) xbt_dict_cursor_free(&cursor); /**If we are currently seeding, we unchoke the peer which has been unchoked the last time.*/ - if (peer->pieces == FILE_PIECES) { + if (has_finished(peer->bitfield)) { connection_t connection; double unchoke_time = MSG_get_clock() + 1; @@ -619,10 +617,12 @@ void update_choked_peers(peer_t peer) i++; } xbt_dict_cursor_free(&cursor); - if ((peer_choosed->interested == 0) || - (peer_choosed->choked_upload == 0)) { + if (peer_choosed == NULL) + THROWF(unknown_error, 0, "A peer should have be selected at this point"); + else if ((peer_choosed->interested == 0) || (peer_choosed->choked_upload == 0)) peer_choosed = NULL; - } + else + XBT_DEBUG("Nothing to do, keep going"); j++; } while (peer_choosed == NULL && j < MAXIMUM_PAIRS); } else { @@ -647,7 +647,7 @@ void update_choked_peers(peer_t peer) if (peer_choked != NULL) { xbt_assert((!peer_choked->choked_upload), "Tries to choked a choked peer"); peer_choked->choked_upload = 1; - xbt_assert((*((int *) key_choked) == peer_choked->id), "WTF !!!"); + xbt_assert((*((int *) key_choked) == peer_choked->id)); update_active_peers_set(peer, peer_choked); XBT_DEBUG("(%d) Sending a CHOKE to %d", peer->id, peer_choked->id); send_choked(peer, peer_choked->mailbox); @@ -672,15 +672,12 @@ void update_interested_after_receive(peer_t peer) char *key; xbt_dict_cursor_t cursor; connection_t connection; - int interested; xbt_dict_foreach(peer->peers, cursor, key, connection) { - interested = 0; if (connection->am_interested != 0) { - xbt_assert(connection->bitfield, "Bitfield not received"); + int interested = 0; //Check if the peer still has a piece we want. - int i; - for (i = 0; i < FILE_PIECES; i++) { - if (connection->bitfield[i] == '1' && peer->bitfield[i] == '0') { + for (int i = 0; i < FILE_PIECES; i++) { + if (peer_has_not_piece(peer, i) && connection_has_piece(connection,i)) { interested = 1; break; } @@ -695,20 +692,18 @@ void update_interested_after_receive(peer_t peer) void update_bitfield_blocks(peer_t peer, int index, int block_index, int block_length) { - int i; xbt_assert((index >= 0 && index <= FILE_PIECES), "Wrong piece."); xbt_assert((block_index >= 0 && block_index <= PIECES_BLOCKS), "Wrong block : %d.", block_index); - for (i = block_index; i < (block_index + block_length); i++) { - peer->bitfield_blocks[index * PIECES_BLOCKS + i] = '1'; + for (int i = block_index; i < (block_index + block_length); i++) { + peer->bitfield_blocks |= (1ULL<<(unsigned int)(index * PIECES_BLOCKS + i)); } } /** Returns if a peer has completed the download of a piece */ int piece_complete(peer_t peer, int index) { - int i; - for (i = 0; i < PIECES_BLOCKS; i++) { - if (peer->bitfield_blocks[index * PIECES_BLOCKS + i] == '0') { + for (int i = 0; i < PIECES_BLOCKS; i++) { + if (!(peer->bitfield_blocks & 1ULL<<(index * PIECES_BLOCKS + i))) { return 0; } } @@ -718,9 +713,8 @@ int piece_complete(peer_t peer, int index) /** Returns the first block that a peer doesn't have in a piece. If the peer has all blocks of the piece, returns -1. */ int get_first_block(peer_t peer, int piece) { - int i; - for (i = 0; i < PIECES_BLOCKS; i++) { - if (peer->bitfield_blocks[piece * PIECES_BLOCKS + i] == '0') { + for (int i = 0; i < PIECES_BLOCKS; i++) { + if (!(peer->bitfield_blocks & 1ULL<<(piece * PIECES_BLOCKS + i))) { return i; } } @@ -730,22 +724,14 @@ int get_first_block(peer_t peer, int piece) /** Indicates if the remote peer has a piece not stored by the local peer */ int is_interested(peer_t peer, connection_t remote_peer) { - xbt_assert(remote_peer->bitfield, "Bitfield not received"); - for (int i = 0; i < FILE_PIECES; i++) { - if (remote_peer->bitfield[i] == '1' && peer->bitfield[i] == '0') { - return 1; - } - } - return 0; + return remote_peer->bitfield & (peer->bitfield^((1<bitfield, "Bitfield not received"); for (int i = 0; i < FILE_PIECES; i++) { - if (remote_peer->bitfield[i] == '1' && peer->bitfield[i] == '0' && - (in_current_pieces(peer, i) == 0)) { + if (peer_has_not_piece(peer, i) && connection_has_piece(remote_peer,i) && peer_is_not_downloading_piece(peer, i)) { return 1; } } @@ -755,13 +741,10 @@ int is_interested_and_free(peer_t peer, connection_t remote_peer) /** Returns a piece that is partially downloaded and stored by the remote peer if any -1 otherwise. */ int partially_downloaded_piece(peer_t peer, connection_t remote_peer) { - xbt_assert(remote_peer->bitfield, "Bitfield not received"); for (int i = 0; i < FILE_PIECES; i++) { - if (remote_peer->bitfield[i] == '1' && peer->bitfield[i] == '0' && - (in_current_pieces(peer, i) == 0)) { - if (get_first_block(peer, i) > 0) - return i; - } + if (peer_has_not_piece(peer, i) && connection_has_piece(remote_peer, i) && peer_is_not_downloading_piece(peer, i) && + get_first_block(peer, i) > 0) + return i; } return -1; } @@ -773,23 +756,14 @@ int partially_downloaded_piece(peer_t peer, connection_t remote_peer) void send_request_to_peer(peer_t peer, connection_t remote_peer, int piece) { remote_peer->current_piece = piece; - int block_index, block_length; - xbt_assert(remote_peer->bitfield, "bitfield not received"); - xbt_assert(remote_peer->bitfield[piece] == '1', "WTF !!!"); - block_index = get_first_block(peer, piece); + xbt_assert(connection_has_piece(remote_peer,piece)); + int block_index = get_first_block(peer, piece); if (block_index != -1) { - block_length = PIECES_BLOCKS - block_index; - block_length = MIN(BLOCKS_REQUESTED, block_length); + int block_length = MIN(BLOCKS_REQUESTED, PIECES_BLOCKS - block_index); send_request(peer, remote_peer->mailbox, piece, block_index, block_length); } } -/** Indicates if a piece is currently being downloaded by the peer. */ -int in_current_pieces(peer_t peer, int piece) -{ - return xbt_dynar_member(peer->current_pieces, &piece); -} - /*********************************************************** * * Low level message functions @@ -842,10 +816,10 @@ void send_handshake_all(peer_t peer) */ void send_handshake(peer_t peer, const char *mailbox) { + XBT_DEBUG("Sending a HANDSHAKE to %s", mailbox); msg_task_t task = task_message_new(MESSAGE_HANDSHAKE, peer->hostname, peer->mailbox, peer->id, task_message_size(MESSAGE_HANDSHAKE)); MSG_task_dsend(task, mailbox, task_message_free); - XBT_DEBUG("Sending a HANDSHAKE to %s", mailbox); } /** Send a "choked" message to a peer. */ @@ -903,7 +877,7 @@ void send_piece(peer_t peer, const char *mailbox, int piece, int block_index, in { XBT_DEBUG("Sending the PIECE %d (%d,%d) to %s", piece, block_index, block_length, mailbox); xbt_assert(piece >= 0, "Tried to send a piece that doesn't exist."); - xbt_assert((peer->bitfield[piece] == '1'), "Tried to send a piece that we doesn't have."); + xbt_assert(!peer_has_not_piece(peer,piece), "Tried to send a piece that we doesn't have."); msg_task_t task = task_message_piece_new(peer->hostname, peer->mailbox, peer->id, piece, block_index, block_length, BLOCK_SIZE); MSG_task_dsend(task, mailbox, task_message_free);