examples/msg/bittorrent/peer.c

   1   /* Copyright (c) 2012. The SimGrid Team.
   2    * All rights reserved.                                                     */
   3
   4 /* This program is free software; you can redistribute it and/or modify it
   5  * under the terms of the license (GNU LGPL) which comes with this package. */
   6 #include "peer.h"
   7 #include "tracker.h"
   8 #include "connection.h"
   9 #include "messages.h"
  10 #include <msg/msg.h>
  11 #include <xbt/RngStream.h>
  12
  13 XBT_LOG_NEW_DEFAULT_CATEGORY(msg_peers, "Messages specific for the peers");
  14
  15 //TODO: Let users change this
  16 /*
  17  * File transfered data
  18  *
  19  * File size: 10 pieces * 5 blocks/piece * 16384 bytes/block = 819200 bytes
  20  */
  21 static int FILE_SIZE = 10 * 5 * 16384;
  22 static int FILE_PIECES = 10;
  23
  24 static int PIECES_BLOCKS = 5;
  25 static int BLOCK_SIZE = 16384;
  26 static int BLOCKS_REQUESTED = 2;
  27
  28 /**
  29  * Peer main function
  30  */
  31 int peer(int argc, char *argv[])
  32 {
  33   s_peer_t peer;
  34   //Check arguments
  35   xbt_assert(argc == 3 || argc == 4, "Wrong number of arguments");
  36   //Build peer object
  37   if (argc == 4) {
  38     peer_init(&peer, atoi(argv[1]), 1);
  39   } else {
  40     peer_init(&peer, atoi(argv[1]), 0);
  41   }
  42   //Retrieve deadline
  43   double deadline = atof(argv[2]);
  44   xbt_assert(deadline > 0, "Wrong deadline supplied");
  45   XBT_INFO("Hi, I'm joining the network with id %d", peer.id);
  46   //Getting peer data from the tracker.
  47   if (get_peers_data(&peer)) {
  48     XBT_DEBUG("Got %d peers from the tracker", xbt_dict_length(peer.peers));
  49     XBT_DEBUG("Here is my current status: %s", peer.bitfield);
  50     peer.begin_receive_time = MSG_get_clock();
  51     if (has_finished(peer.bitfield)) {
  52       peer.pieces = FILE_PIECES;
  53       send_handshake_all(&peer);
  54       seed_loop(&peer, deadline);
  55     } else {
  56       leech_loop(&peer, deadline);
  57       seed_loop(&peer, deadline);
  58     }
  59   } else {
  60     XBT_INFO("Couldn't contact the tracker.");
  61   }
  62
  63   XBT_INFO("Here is my current status: %s", peer.bitfield);
  64   if (peer.comm_received) {
  65     MSG_comm_destroy(peer.comm_received);
  66   }
  67
  68   peer_free(&peer);
  69   return 0;
  70 }
  71
  72 /**
  73  * Peer main loop when it is leeching.
  74  * @param peer peer data
  75  * @param deadline time at which the peer has to leave
  76  */
  77 void leech_loop(peer_t peer, double deadline)
  78 {
  79   double next_choked_update = MSG_get_clock() + UPDATE_CHOKED_INTERVAL;
  80   XBT_DEBUG("Start downloading.");
  81   /*
  82    * Send a "handshake" message to all the peers it got
  83    * (since it couldn't have gotten more than 50 peers)
  84    */
  85   send_handshake_all(peer);
  86   //Wait for at least one "bitfield" message.
  87   wait_for_pieces(peer, deadline);
  88   XBT_DEBUG("Starting main leech loop");
  89   while (MSG_get_clock() < deadline && peer->pieces < FILE_PIECES) {
  90     if (peer->comm_received == NULL) {
  91       peer->task_received = NULL;
  92       peer->comm_received = MSG_task_irecv(&peer->task_received, peer->mailbox);
  93     }
  94     if (MSG_comm_test(peer->comm_received)) {
  95       msg_error_t status = MSG_comm_get_status(peer->comm_received);
  96       MSG_comm_destroy(peer->comm_received);
  97       peer->comm_received = NULL;
  98       if (status == MSG_OK) {
  99         handle_message(peer, peer->task_received);
 100       }
 101     } else {
 102       handle_pending_sends(peer);
 103       if (peer->current_piece != -1) {
 104         send_interested_to_peers(peer);
 105       } else {
 106         //If the current interested pieces is < MAX
 107         if (peer->pieces_requested < MAX_PIECES) {
 108           update_current_piece(peer);
 109         }
 110       }
 111       //We don't execute the choke algorithm if we don't already have a piece
 112       if (MSG_get_clock() >= next_choked_update && peer->pieces > 0) {
 113         update_choked_peers(peer);
 114         next_choked_update += UPDATE_CHOKED_INTERVAL;
 115       } else {
 116         MSG_process_sleep(1);
 117       }
 118     }
 119   }
 120 }
 121
 122 /**
 123  * Peer main loop when it is seeding
 124  * @param peer peer data
 125  * @param deadline time when the peer will leave
 126  */
 127 void seed_loop(peer_t peer, double deadline)
 128 {
 129   double next_choked_update = MSG_get_clock() + UPDATE_CHOKED_INTERVAL;
 130   XBT_DEBUG("Start seeding.");
 131   //start the main seed loop
 132   while (MSG_get_clock() < deadline) {
 133     if (peer->comm_received == NULL) {
 134       peer->task_received = NULL;
 135       peer->comm_received = MSG_task_irecv(&peer->task_received, peer->mailbox);
 136     }
 137     if (MSG_comm_test(peer->comm_received)) {
 138       msg_error_t status = MSG_comm_get_status(peer->comm_received);
 139       MSG_comm_destroy(peer->comm_received);
 140       peer->comm_received = NULL;
 141       if (status == MSG_OK) {
 142         handle_message(peer, peer->task_received);
 143       }
 144     } else {
 145       if (MSG_get_clock() >= next_choked_update) {
 146         update_choked_peers(peer);
 147         //TODO: Change the choked peer algorithm when seeding.
 148         next_choked_update += UPDATE_CHOKED_INTERVAL;
 149       } else {
 150         MSG_process_sleep(1);
 151       }
 152     }
 153   }
 154 }
 155
 156 /**
 157  * Retrieves the peer list from the tracker
 158  * @param peer current peer data
 159  */
 160 int get_peers_data(peer_t peer)
 161 {
 162   int success = 0, send_success = 0;
 163   double timeout = MSG_get_clock() + GET_PEERS_TIMEOUT;
 164   //Build the task to send to the tracker
 165   tracker_task_data_t data =
 166       tracker_task_data_new(MSG_host_get_name(MSG_host_self()),
 167                             peer->mailbox_tracker, peer->id, 0, 0, FILE_SIZE);
 168   //Build the task to send.
 169   msg_task_t task_send = MSG_task_create(NULL, 0, TRACKER_COMM_SIZE, data);
 170   msg_task_t task_received = NULL;
 171   msg_comm_t comm_received;
 172   while (!send_success && MSG_get_clock() < timeout) {
 173     XBT_DEBUG("Sending a peer request to the tracker.");
 174     msg_error_t status =
 175         MSG_task_send_with_timeout(task_send, TRACKER_MAILBOX,
 176                                    GET_PEERS_TIMEOUT);
 177     if (status == MSG_OK) {
 178       send_success = 1;
 179     }
 180   }
 181   while (!success && MSG_get_clock() < timeout) {
 182     comm_received = MSG_task_irecv(&task_received, peer->mailbox_tracker);
 183     msg_error_t status = MSG_comm_wait(comm_received, GET_PEERS_TIMEOUT);
 184     if (status == MSG_OK) {
 185       tracker_task_data_t data = MSG_task_get_data(task_received);
 186       unsigned i;
 187       int peer_id;
 188       //Add the peers the tracker gave us to our peer list.
 189       xbt_dynar_foreach(data->peers, i, peer_id) {
 190         if (peer_id != peer->id)
 191           xbt_dict_set_ext(peer->peers, (char *) &peer_id, sizeof(int),
 192                            connection_new(peer_id), NULL);
 193       }
 194       success = 1;
 195       //free the communication and the task
 196       MSG_comm_destroy(comm_received);
 197       tracker_task_data_free(data);
 198       MSG_task_destroy(task_received);
 199       comm_received = NULL;
 200     }
 201   }
 202
 203   return success;
 204 }
 205
 206 /**
 207  * Initialize the peer data.
 208  * @param peer peer data
 209  * @param id id of the peer to take in the network
 210  * @param seed indicates if the peer is a seed.
 211  */
 212 void peer_init(peer_t peer, int id, int seed)
 213 {
 214   peer->id = id;
 215   sprintf(peer->mailbox, "%d", id);
 216   sprintf(peer->mailbox_tracker, "tracker_%d", id);
 217   peer->peers = xbt_dict_new();
 218   peer->active_peers = xbt_dict_new();
 219   peer->hostname = MSG_host_get_name(MSG_host_self());
 220
 221   peer->bitfield = xbt_new(char, FILE_PIECES + 1);
 222   peer->bitfield_blocks = xbt_new(char, (FILE_PIECES) * (PIECES_BLOCKS) + 1);
 223   if (seed) {
 224     memset(peer->bitfield, '1', sizeof(char) * (FILE_PIECES + 1));
 225     memset(peer->bitfield_blocks, '1',
 226            sizeof(char) * FILE_PIECES * (PIECES_BLOCKS));
 227   } else {
 228     memset(peer->bitfield, '0', sizeof(char) * (FILE_PIECES + 1));
 229     memset(peer->bitfield_blocks, '0',
 230            sizeof(char) * FILE_PIECES * (PIECES_BLOCKS));
 231   }
 232
 233   peer->bitfield[FILE_PIECES] = '\0';
 234   peer->pieces = 0;
 235
 236   peer->pieces_count = xbt_new0(short, FILE_PIECES);
 237   peer->pieces_requested = 0;
 238
 239   peer->current_pieces = xbt_dynar_new(sizeof(int), NULL);
 240   peer->current_piece = -1;
 241
 242   peer->stream = MSG_host_get_data(MSG_host_self());
 243   peer->comm_received = NULL;
 244
 245   peer->round = 0;
 246
 247   peer->pending_sends = xbt_dynar_new(sizeof(msg_comm_t), NULL);
 248 }
 249
 250 /**
 251  * Destroys a poor peer object.
 252  */
 253 void peer_free(peer_t peer)
 254 {
 255   char *key;
 256   connection_t connection;
 257   xbt_dict_cursor_t cursor;
 258   xbt_dict_foreach(peer->peers, cursor, key, connection) {
 259     connection_free(connection);
 260   }
 261   xbt_dict_free(&peer->peers);
 262   xbt_dict_free(&peer->active_peers);
 263   xbt_dynar_free(&peer->current_pieces);
 264   xbt_dynar_free(&peer->pending_sends);
 265   xbt_free(peer->pieces_count);
 266   xbt_free(peer->bitfield);
 267   xbt_free(peer->bitfield_blocks);
 268 }
 269
 270 /**
 271  * Returns if a peer has finished downloading the file
 272  * @param bitfield peer bitfield
 273  */
 274 int has_finished(char *bitfield)
 275 {
 276   return ((memchr(bitfield, '0', sizeof(char) * FILE_PIECES) == NULL) ? 1 : 0);
 277 }
 278
 279 /**
 280  * Handle pending sends and remove those which are done
 281  * @param peer Peer data
 282  */
 283 void handle_pending_sends(peer_t peer)
 284 {
 285   int index;
 286
 287   while ((index = MSG_comm_testany(peer->pending_sends)) != -1) {
 288     msg_comm_t comm_send = xbt_dynar_get_as(peer->pending_sends, index, msg_comm_t);
 289     int status = MSG_comm_get_status(comm_send);
 290     xbt_dynar_remove_at(peer->pending_sends, index, &comm_send);
 291     XBT_DEBUG("Communication %p is finished with status %d, dynar size is now %lu", comm_send, status, xbt_dynar_length(peer->pending_sends));
 292
 293     msg_task_t task = MSG_comm_get_task(comm_send);
 294     MSG_comm_destroy(comm_send);
 295
 296     if (status != MSG_OK) {
 297       task_message_free(task);
 298     }
 299   }
 300 }
 301
 302 /**
 303  * Handle a received message sent by another peer
 304  * @param peer Peer data
 305  * @param task task received.
 306  */
 307 void handle_message(peer_t peer, msg_task_t task)
 308 {
 309   message_t message = MSG_task_get_data(task);
 310   connection_t remote_peer;
 311   remote_peer =
 312       xbt_dict_get_or_null_ext(peer->peers, (char *) &message->peer_id,
 313                                sizeof(int));
 314   switch (message->type) {
 315
 316   case MESSAGE_HANDSHAKE:
 317     XBT_DEBUG("Received a HANDSHAKE from %s (%s)", message->mailbox,
 318               message->issuer_host_name);
 319     //Check if the peer is in our connection list.
 320     if (!remote_peer) {
 321       xbt_dict_set_ext(peer->peers, (char *) &message->peer_id, sizeof(int),
 322                        connection_new(message->peer_id), NULL);
 323       send_handshake(peer, message->mailbox);
 324     }
 325     //Send our bitfield to the peer
 326     send_bitfield(peer, message->mailbox);
 327     break;
 328   case MESSAGE_BITFIELD:
 329     XBT_DEBUG("Recieved a BITFIELD message from %s (%s)", message->mailbox,
 330               message->issuer_host_name);
 331     //Update the pieces list
 332     update_pieces_count_from_bitfield(peer, message->bitfield);
 333     //Store the bitfield
 334     remote_peer->bitfield = xbt_strdup(message->bitfield);
 335     //Update the current piece
 336     if (peer->current_piece == -1 && peer->pieces < FILE_PIECES) {
 337       update_current_piece(peer);
 338     }
 339     break;
 340   case MESSAGE_INTERESTED:
 341     XBT_DEBUG("Recieved an INTERESTED message from %s (%s)", message->mailbox,
 342               message->issuer_host_name);
 343     xbt_assert((remote_peer != NULL),
 344                "A non-in-our-list peer has sent us a message. WTH ?");
 345     //Update the interested state of the peer.
 346     remote_peer->interested = 1;
 347     break;
 348   case MESSAGE_NOTINTERESTED:
 349     XBT_DEBUG("Received a NOTINTERESTED message from %s (%s)", message->mailbox,
 350               message->issuer_host_name);
 351     xbt_assert((remote_peer != NULL),
 352                "A non-in-our-list peer has sent us a message. WTH ?");
 353     remote_peer->interested = 0;
 354     break;
 355   case MESSAGE_UNCHOKE:
 356     xbt_assert((remote_peer != NULL),
 357                "A non-in-our-list peer has sent us a message. WTH ?");
 358     XBT_DEBUG("Received a UNCHOKE message from %s (%s)", message->mailbox,
 359               message->issuer_host_name);
 360     remote_peer->choked_download = 0;
 361     xbt_dict_set_ext(peer->active_peers, (char *) &message->peer_id,
 362                      sizeof(int), remote_peer, NULL);
 363     //Send requests to the peer, since it has unchoked us
 364     send_requests_to_peer(peer, remote_peer);
 365     break;
 366   case MESSAGE_CHOKE:
 367     xbt_assert((remote_peer != NULL),
 368                "A non-in-our-list peer has sent us a message. WTH ?");
 369     XBT_DEBUG("Received a CHOKE message from %s (%s)", message->mailbox,
 370               message->issuer_host_name);
 371     remote_peer->choked_download = 1;
 372     xbt_ex_t e;
 373     TRY {
 374       xbt_dict_remove_ext(peer->active_peers, (char *) &message->peer_id,
 375                           sizeof(int));
 376     }
 377     CATCH(e) {
 378       xbt_ex_free(e);
 379     }
 380     break;
 381   case MESSAGE_HAVE:
 382     XBT_DEBUG("Received a HAVE message from %s (%s) of piece %d",
 383               message->mailbox, message->issuer_host_name, message->index);
 384     xbt_assert((message->index >= 0
 385                 && message->index < FILE_PIECES),
 386                "Wrong HAVE message received");
 387     if (remote_peer->bitfield == NULL)
 388       return;
 389     remote_peer->bitfield[message->index] = '1';
 390     peer->pieces_count[message->index]++;
 391     //If the piece is in our pieces, we tell the peer that we are interested.
 392     if (!remote_peer->am_interested && in_current_pieces(peer, message->index)) {
 393       remote_peer->am_interested = 1;
 394       send_interested(peer, remote_peer->mailbox);
 395     }
 396     break;
 397   case MESSAGE_REQUEST:
 398     xbt_assert((message->index >= 0
 399                 && message->index < FILE_PIECES), "Wrong request received");
 400     if (!remote_peer->choked_upload) {
 401       XBT_DEBUG("Received a REQUEST from %s (%s) for %d (%d,%d)",
 402                 message->mailbox, message->issuer_host_name, message->peer_id,
 403                 message->block_index,
 404                 message->block_index + message->block_length);
 405       if (peer->bitfield[message->index] == '1') {
 406         send_piece(peer, message->mailbox, message->index, 0,
 407                    message->block_index, message->block_length);
 408       }
 409     } else {
 410       XBT_DEBUG("Received a REQUEST from %s (%s) for %d but he is choked.",
 411                 message->mailbox, message->issuer_host_name, message->peer_id);
 412     }
 413     break;
 414   case MESSAGE_PIECE:
 415     xbt_assert((message->index >= 0
 416                 && message->index < FILE_PIECES), "Wrong piece received");
 417     //TODO: Execute à computation.
 418     if (message->stalled) {
 419       XBT_DEBUG("The received piece %d from %s (%s) is STALLED", message->index,
 420                 message->mailbox, message->issuer_host_name);
 421     } else {
 422       XBT_DEBUG("Received piece %d (%d,%d) from %s (%s)", message->index,
 423                 message->block_index,
 424                 message->block_index + message->block_length, message->mailbox,
 425                 message->issuer_host_name);
 426       if (peer->bitfield[message->index] == '0') {
 427         update_bitfield_blocks(peer, message->index, message->block_index,
 428                                message->block_length);
 429         if (piece_complete(peer, message->index)) {
 430           peer->pieces_requested--;
 431           //Removing the piece from our piece list
 432           unsigned i;
 433           int piece_index = -1, piece;
 434           xbt_dynar_foreach(peer->current_pieces, i, piece) {
 435             if (piece == message->index) {
 436               piece_index = i;
 437               break;
 438             }
 439           }
 440           xbt_assert(piece_index != -1, "Received an incorrect piece");
 441           xbt_dynar_remove_at(peer->current_pieces, piece_index, NULL);
 442           //Setting the fact that we have the piece
 443           peer->bitfield[message->index] = '1';
 444           peer->pieces++;
 445           XBT_DEBUG("My status is now %s", peer->bitfield);
 446           //Sending the information to all the peers we are connected to
 447           send_have(peer, message->index);
 448           //sending UNINTERSTED to peers that doesn't have what we want.
 449           update_interested_after_receive(peer);
 450         }
 451       } else {
 452         XBT_DEBUG("However, we already have it");
 453       }
 454     }
 455     break;
 456   case MESSAGE_CANCEL:
 457     break;
 458   }
 459   //Update the peer speed.
 460   if (remote_peer) {
 461     connection_add_speed_value(remote_peer,
 462                                1.0 / (MSG_get_clock() -
 463                                       peer->begin_receive_time));
 464   }
 465   peer->begin_receive_time = MSG_get_clock();
 466
 467   task_message_free(task);
 468 }
 469
 470 /**
 471  * Wait for the node to receive interesting bitfield messages (ie: non empty)
 472  * to be received
 473  * @param deadline peer deadline
 474  * @param peer peer data
 475  */
 476 void wait_for_pieces(peer_t peer, double deadline)
 477 {
 478   int finished = 0;
 479   while (MSG_get_clock() < deadline && !finished) {
 480     if (peer->comm_received == NULL) {
 481       peer->task_received = NULL;
 482       peer->comm_received = MSG_task_irecv(&peer->task_received, peer->mailbox);
 483     }
 484     msg_error_t status = MSG_comm_wait(peer->comm_received, TIMEOUT_MESSAGE);
 485     //free the comm already, we don't need it anymore
 486     MSG_comm_destroy(peer->comm_received);
 487     peer->comm_received = NULL;
 488     if (status == MSG_OK) {
 489       MSG_task_get_data(peer->task_received);
 490       handle_message(peer, peer->task_received);
 491       if (peer->current_piece != -1) {
 492         finished = 1;
 493       }
 494     }
 495   }
 496 }
 497
 498 /**
 499  * Updates the list of who has a piece from a bitfield
 500  * @param peer peer we want to update the list
 501  * @param bitfield bitfield
 502  */
 503 void update_pieces_count_from_bitfield(peer_t peer, char *bitfield)
 504 {
 505   int i;
 506   for (i = 0; i < FILE_PIECES; i++) {
 507     if (bitfield[i] == '1') {
 508       peer->pieces_count[i]++;
 509     }
 510   }
 511 }
 512
 513 /**
 514  * Update the piece the peer is currently interested in.
 515  * There is two cases (as described in "Bittorrent Architecture Protocol", Ryan Toole :
 516  * If the peer has less than 3 pieces, he chooses a piece at random.
 517  * If the peer has more than pieces, he downloads the pieces that are the less
 518  * replicated
 519  */
 520 void update_current_piece(peer_t peer)
 521 {
 522   if (xbt_dynar_length(peer->current_pieces) >= (FILE_PIECES - peer->pieces)) {
 523     return;
 524   }
 525   if (1 || peer->pieces < 3) {
 526     int i = 0;
 527     do {
 528       peer->current_piece =
 529           RngStream_RandInt(peer->stream, 0, FILE_PIECES - 1);;
 530       i++;
 531     } while (!
 532              (peer->bitfield[peer->current_piece] == '0'
 533               && !in_current_pieces(peer, peer->current_piece)));
 534   } else {
 535     //Trivial min algorithm.
 536     int i, min_id = -1;
 537     short min = -1;
 538     for (i = 0; i < FILE_PIECES; i++) {
 539       if (peer->bitfield[i] == '0') {
 540         min = peer->pieces_count[i];
 541         min_id = i;
 542         break;
 543       }
 544     }
 545     xbt_assert((min > -1), "Couldn't find a minimum");
 546     for (i = 1; i < FILE_PIECES; i++) {
 547       if (peer->pieces_count[i] < min && peer->bitfield[i] == '0') {
 548         min = peer->pieces_count[i];
 549         min_id = i;
 550       }
 551     }
 552     peer->current_piece = min_id;
 553   }
 554   xbt_dynar_push_as(peer->current_pieces, int, peer->current_piece);
 555   XBT_DEBUG("New interested piece: %d", peer->current_piece);
 556   xbt_assert((peer->current_piece >= 0 && peer->current_piece < FILE_PIECES),
 557              "Peer want to retrieve a piece that doesn't exist.");
 558 }
 559
 560 /**
 561  * Update the list of current choked and unchoked peers, using the
 562  * choke algorithm
 563  * @param peer the current peer
 564  */
 565 void update_choked_peers(peer_t peer)
 566 {
 567   //update the current round
 568   peer->round = (peer->round + 1) % 3;
 569   char *key;
 570   connection_t peer_choosed = NULL;
 571   //remove a peer from the list
 572   xbt_dict_cursor_t cursor = NULL;
 573   xbt_dict_cursor_first(peer->active_peers, &cursor);
 574   if (xbt_dict_length(peer->active_peers) > 0) {
 575     key = xbt_dict_cursor_get_key(cursor);
 576     connection_t peer_choked = xbt_dict_cursor_get_data(cursor);
 577     if (peer_choked) {
 578       send_choked(peer, peer_choked->mailbox);
 579       peer_choked->choked_upload = 1;
 580     }
 581     xbt_dict_remove_ext(peer->active_peers, key, sizeof(int));
 582   }
 583   xbt_dict_cursor_free(&cursor);
 584
 585         /**
 586          * If we are currently seeding, we unchoke the peer which has
 587          * been unchoke the least time.
 588          */
 589   if (peer->pieces == FILE_PIECES) {
 590     connection_t connection;
 591     double unchoke_time = MSG_get_clock() + 1;
 592
 593     xbt_dict_foreach(peer->peers, cursor, key, connection) {
 594       if (connection->last_unchoke < unchoke_time && connection->interested) {
 595         unchoke_time = connection->last_unchoke;
 596         peer_choosed = connection;
 597       }
 598     }
 599   } else {
 600     //Random optimistic unchoking
 601     if (peer->round == 0) {
 602       int j = 0;
 603       do {
 604         //We choose a random peer to unchoke.
 605         int id_chosen =
 606             RngStream_RandInt(peer->stream, 0,
 607                               xbt_dict_length(peer->peers) - 1);
 608         int i = 0;
 609         connection_t connection;
 610         xbt_dict_foreach(peer->peers, cursor, key, connection) {
 611           if (i == id_chosen) {
 612             peer_choosed = connection;
 613             break;
 614           }
 615           i++;
 616         }
 617         xbt_dict_cursor_free(&cursor);
 618         if (peer_choosed->interested == 0) {
 619           peer_choosed = NULL;
 620         }
 621         j++;
 622       } while (peer_choosed == NULL && j < MAXIMUM_PAIRS);
 623     } else {
 624       //Use the "fastest download" policy.
 625       connection_t connection;
 626       double fastest_speed = 0.0;
 627       xbt_dict_foreach(peer->peers, cursor, key, connection) {
 628         if (connection->peer_speed > fastest_speed && connection->choked_upload
 629             && connection->interested) {
 630           peer_choosed = connection;
 631           fastest_speed = connection->peer_speed;
 632         }
 633       }
 634     }
 635
 636   }
 637   if (peer_choosed != NULL) {
 638     peer_choosed->choked_upload = 0;
 639     xbt_dict_set_ext(peer->active_peers, (char *) &peer_choosed->id,
 640                      sizeof(int), peer_choosed, NULL);
 641     peer_choosed->last_unchoke = MSG_get_clock();
 642     send_unchoked(peer, peer_choosed->mailbox);
 643   }
 644
 645 }
 646
 647 /**
 648  * Updates our "interested" state about peers: send "not interested" to peers
 649  * that don't have any more pieces we want.
 650  * @param peer our peer data
 651  */
 652 void update_interested_after_receive(peer_t peer)
 653 {
 654   char *key;
 655   xbt_dict_cursor_t cursor;
 656   connection_t connection;
 657   unsigned cpt;
 658   int interested, piece;
 659   xbt_dict_foreach(peer->peers, cursor, key, connection) {
 660     interested = 0;
 661     if (connection->am_interested) {
 662       //Check if the peer still has a piece we want.
 663       xbt_dynar_foreach(peer->current_pieces, cpt, piece) {
 664         xbt_assert((piece >= 0), "Wrong piece.");
 665         if (connection->bitfield && connection->bitfield[piece] == '1') {
 666           interested = 1;
 667           break;
 668         }
 669       }
 670       if (!interested) {
 671         connection->am_interested = 0;
 672         send_notinterested(peer, connection->mailbox);
 673       }
 674     }
 675   }
 676 }
 677
 678 void update_bitfield_blocks(peer_t peer, int index, int block_index,
 679                             int block_length)
 680 {
 681   int i;
 682   xbt_assert((index >= 0 && index <= FILE_PIECES), "Wrong piece.");
 683   xbt_assert((block_index >= 0
 684               && block_index <= PIECES_BLOCKS), "Wrong block : %d.",
 685              block_index);
 686   for (i = block_index; i < (block_index + block_length); i++) {
 687     peer->bitfield_blocks[index * PIECES_BLOCKS + i] = '1';
 688   }
 689 }
 690
 691 /**
 692  * Returns if a peer has completed the download of a piece
 693  */
 694 int piece_complete(peer_t peer, int index)
 695 {
 696   int i;
 697   for (i = 0; i < PIECES_BLOCKS; i++) {
 698     if (peer->bitfield_blocks[index * PIECES_BLOCKS + i] == '0') {
 699       return 0;
 700     }
 701   }
 702   return 1;
 703 }
 704
 705 /**
 706  * Returns the first block that a peer doesn't have in a piece
 707  */
 708 int get_first_block(peer_t peer, int piece)
 709 {
 710   int i;
 711   for (i = 0; i < PIECES_BLOCKS; i++) {
 712     if (peer->bitfield_blocks[piece * PIECES_BLOCKS + i] == '0') {
 713       return i;
 714     }
 715   }
 716   return -1;
 717 }
 718
 719 /**
 720  * Send request messages to a peer that have unchoked us
 721  * @param peer peer
 722  * @param remote_peer peer data to the peer we want to send the request
 723  */
 724 void send_requests_to_peer(peer_t peer, connection_t remote_peer)
 725 {
 726   unsigned i;
 727   int piece, block_index, block_length;
 728   xbt_dynar_foreach(peer->current_pieces, i, piece) {
 729     if (remote_peer->bitfield && remote_peer->bitfield[piece] == '1') {
 730       block_index = get_first_block(peer, piece);
 731       if (block_index != -1) {
 732         block_length = PIECES_BLOCKS - block_index;
 733         block_length = min(BLOCKS_REQUESTED, block_length);
 734         send_request(peer, remote_peer->mailbox, piece, block_index,
 735                      block_length);
 736         break;
 737       }
 738     }
 739   }
 740 }
 741
 742 /**
 743  * Find the peers that have the current interested piece and send them
 744  * the "interested" message
 745  */
 746 void send_interested_to_peers(peer_t peer)
 747 {
 748   char *key;
 749   xbt_dict_cursor_t cursor = NULL;
 750   connection_t connection;
 751   xbt_assert((peer->current_piece != -1),
 752              "Tried to send a interested message wheras the current_piece is -1");
 753   xbt_dict_foreach(peer->peers, cursor, key, connection) {
 754     if (connection->bitfield
 755         && connection->bitfield[peer->current_piece] == '1') {
 756       connection->am_interested = 1;
 757       msg_task_t task =
 758           task_message_new(MESSAGE_INTERESTED, peer->hostname, peer->mailbox,
 759                            peer->id, task_message_size(MESSAGE_INTERESTED));
 760       MSG_task_dsend(task, connection->mailbox, task_message_free);
 761       XBT_DEBUG("Send INTERESTED to %s", connection->mailbox);
 762     }
 763   }
 764   peer->current_piece = -1;
 765   peer->pieces_requested++;
 766 }
 767
 768 /**
 769  * Send a "interested" message to a peer
 770  * @param peer peer data
 771  * @param mailbox destination mailbox
 772  */
 773 void send_interested(peer_t peer, const char *mailbox)
 774 {
 775   msg_task_t task =
 776       task_message_new(MESSAGE_INTERESTED, peer->hostname, peer->mailbox,
 777                        peer->id, task_message_size(MESSAGE_INTERESTED));
 778   MSG_task_dsend(task, mailbox, task_message_free);
 779   XBT_DEBUG("Sending INTERESTED to %s", mailbox);
 780
 781 }
 782
 783 /**
 784  * Send a "not interested" message to a peer
 785  * @param peer peer data
 786  * @param mailbox destination mailbox
 787  */
 788 void send_notinterested(peer_t peer, const char *mailbox)
 789 {
 790   msg_task_t task =
 791       task_message_new(MESSAGE_NOTINTERESTED, peer->hostname, peer->mailbox,
 792                        peer->id, task_message_size(MESSAGE_NOTINTERESTED));
 793   MSG_task_dsend(task, mailbox, task_message_free);
 794   XBT_DEBUG("Sending NOTINTERESTED to %s", mailbox);
 795
 796 }
 797
 798 /**
 799  * Send a handshake message to all the peers the peer has.
 800  * @param peer peer data
 801  */
 802 void send_handshake_all(peer_t peer)
 803 {
 804   connection_t remote_peer;
 805   xbt_dict_cursor_t cursor = NULL;
 806   char *key;
 807   xbt_dict_foreach(peer->peers, cursor, key, remote_peer) {
 808     msg_task_t task =
 809         task_message_new(MESSAGE_HANDSHAKE, peer->hostname, peer->mailbox,
 810                          peer->id, task_message_size(MESSAGE_HANDSHAKE));
 811     MSG_task_dsend(task, remote_peer->mailbox, task_message_free);
 812     XBT_DEBUG("Sending a HANDSHAKE to %s", remote_peer->mailbox);
 813   }
 814 }
 815
 816 /**
 817  * Send a "handshake" message to an user
 818  * @param peer peer data
 819  * @param mailbox mailbox where to we send the message
 820  */
 821 void send_handshake(peer_t peer, const char *mailbox)
 822 {
 823   msg_task_t task =
 824       task_message_new(MESSAGE_HANDSHAKE, peer->hostname, peer->mailbox,
 825                        peer->id, task_message_size(MESSAGE_HANDSHAKE));
 826   MSG_task_dsend(task, mailbox, task_message_free);
 827   XBT_DEBUG("Sending a HANDSHAKE to %s", mailbox);
 828 }
 829
 830 /**
 831  * Send a "choked" message to a peer.
 832  */
 833 void send_choked(peer_t peer, const char *mailbox)
 834 {
 835   XBT_DEBUG("Sending a CHOKE to %s", mailbox);
 836   msg_task_t task =
 837       task_message_new(MESSAGE_CHOKE, peer->hostname, peer->mailbox,
 838                        peer->id, task_message_size(MESSAGE_CHOKE));
 839   MSG_task_dsend(task, mailbox, task_message_free);
 840 }
 841
 842 /**
 843  * Send a "unchoked" message to a peer
 844  */
 845 void send_unchoked(peer_t peer, const char *mailbox)
 846 {
 847   XBT_DEBUG("Sending a UNCHOKE to %s", mailbox);
 848   msg_task_t task =
 849       task_message_new(MESSAGE_UNCHOKE, peer->hostname, peer->mailbox,
 850                        peer->id, task_message_size(MESSAGE_UNCHOKE));
 851   MSG_task_dsend(task, mailbox, task_message_free);
 852 }
 853
 854 /**
 855  * Send a "HAVE" message to all peers we are connected to
 856  */
 857 void send_have(peer_t peer, int piece)
 858 {
 859   XBT_DEBUG("Sending HAVE message to all my peers");
 860   connection_t remote_peer;
 861   xbt_dict_cursor_t cursor = NULL;
 862   char *key;
 863   xbt_dict_foreach(peer->peers, cursor, key, remote_peer) {
 864     msg_task_t task =
 865         task_message_index_new(MESSAGE_HAVE, peer->hostname, peer->mailbox,
 866                                peer->id, piece, task_message_size(MESSAGE_HAVE));
 867     MSG_task_dsend(task, remote_peer->mailbox, task_message_free);
 868   }
 869 }
 870
 871 /**
 872  * Send a bitfield message to all the peers the peer has.
 873  * @param peer peer data
 874  */
 875 void send_bitfield(peer_t peer, const char *mailbox)
 876 {
 877   XBT_DEBUG("Sending a BITFIELD to %s", mailbox);
 878   msg_task_t task =
 879       task_message_bitfield_new(peer->hostname, peer->mailbox, peer->id,
 880                                 peer->bitfield, FILE_PIECES);
 881   //Async send and append to pending sends
 882   msg_comm_t comm = MSG_task_isend(task, mailbox);
 883   xbt_dynar_push(peer->pending_sends, &comm);
 884 }
 885
 886 /**
 887  * Send a "request" message to a pair, containing a request for a piece
 888  */
 889 void send_request(peer_t peer, const char *mailbox, int piece, int block_index,
 890                   int block_length)
 891 {
 892   XBT_DEBUG("Sending a REQUEST to %s for piece %d (%d,%d)", mailbox, piece,
 893             block_index, block_length);
 894   msg_task_t task =
 895       task_message_request_new(peer->hostname, peer->mailbox, peer->id, piece,
 896                                block_index, block_length);
 897   MSG_task_dsend(task, mailbox, task_message_free);
 898 }
 899
 900 /**
 901  * Send a "piece" message to a pair, containing a piece of the file
 902  */
 903 void send_piece(peer_t peer, const char *mailbox, int piece, int stalled,
 904                 int block_index, int block_length)
 905 {
 906   XBT_DEBUG("Sending the PIECE %d (%d,%d) to %s", piece, block_index,
 907             block_length, mailbox);
 908   xbt_assert(piece >= 0, "Tried to send a piece that doesn't exist.");
 909   xbt_assert((peer->bitfield[piece] == '1'),
 910              "Tried to send a piece that we doesn't have.");
 911   msg_task_t task =
 912       task_message_piece_new(peer->hostname, peer->mailbox, peer->id, piece,
 913                              stalled, block_index, block_length, BLOCK_SIZE);
 914   MSG_task_dsend(task, mailbox, task_message_free);
 915 }
 916
 917 int in_current_pieces(peer_t peer, int piece)
 918 {
 919   unsigned i;
 920   int is_in = 0, peer_piece;
 921   xbt_dynar_foreach(peer->current_pieces, i, peer_piece) {
 922     if (peer_piece == piece) {
 923       is_in = 1;
 924       break;
 925     }
 926   }
 927   return is_in;
 928 }