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 = RngStream_CreateStream("");
 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   RngStream_DeleteStream(&peer->stream);
 270 }
 271
 272 /**
 273  * Returns if a peer has finished downloading the file
 274  * @param bitfield peer bitfield
 275  */
 276 int has_finished(char *bitfield)
 277 {
 278   return ((memchr(bitfield, '0', sizeof(char) * FILE_PIECES) == NULL) ? 1 : 0);
 279 }
 280
 281 /**
 282  * Handle pending sends and remove those which are done
 283  * @param peer Peer data
 284  */
 285 void handle_pending_sends(peer_t peer)
 286 {
 287   int index;
 288
 289   while ((index = MSG_comm_testany(peer->pending_sends)) != -1) {
 290     msg_comm_t comm_send = xbt_dynar_get_as(peer->pending_sends, index, msg_comm_t);
 291     int status = MSG_comm_get_status(comm_send);
 292     xbt_dynar_remove_at(peer->pending_sends, index, &comm_send);
 293     XBT_DEBUG("Communication %p is finished with status %d, dynar size is now %lu", comm_send, status, xbt_dynar_length(peer->pending_sends));
 294
 295     msg_task_t task = MSG_comm_get_task(comm_send);
 296     MSG_comm_destroy(comm_send);
 297
 298     if (status != MSG_OK) {
 299       task_message_free(task);
 300     }
 301   }
 302 }
 303
 304 /**
 305  * Handle a received message sent by another peer
 306  * @param peer Peer data
 307  * @param task task received.
 308  */
 309 void handle_message(peer_t peer, msg_task_t task)
 310 {
 311   message_t message = MSG_task_get_data(task);
 312   connection_t remote_peer;
 313   remote_peer =
 314       xbt_dict_get_or_null_ext(peer->peers, (char *) &message->peer_id,
 315                                sizeof(int));
 316   switch (message->type) {
 317
 318   case MESSAGE_HANDSHAKE:
 319     XBT_DEBUG("Received a HANDSHAKE from %s (%s)", message->mailbox,
 320               message->issuer_host_name);
 321     //Check if the peer is in our connection list.
 322     if (!remote_peer) {
 323       xbt_dict_set_ext(peer->peers, (char *) &message->peer_id, sizeof(int),
 324                        connection_new(message->peer_id), NULL);
 325       send_handshake(peer, message->mailbox);
 326     }
 327     //Send our bitfield to the peer
 328     send_bitfield(peer, message->mailbox);
 329     break;
 330   case MESSAGE_BITFIELD:
 331     XBT_DEBUG("Recieved a BITFIELD message from %s (%s)", message->mailbox,
 332               message->issuer_host_name);
 333     //Update the pieces list
 334     update_pieces_count_from_bitfield(peer, message->bitfield);
 335     //Store the bitfield
 336     remote_peer->bitfield = xbt_strdup(message->bitfield);
 337     //Update the current piece
 338     if (peer->current_piece == -1 && peer->pieces < FILE_PIECES) {
 339       update_current_piece(peer);
 340     }
 341     break;
 342   case MESSAGE_INTERESTED:
 343     XBT_DEBUG("Recieved an INTERESTED message from %s (%s)", message->mailbox,
 344               message->issuer_host_name);
 345     xbt_assert((remote_peer != NULL),
 346                "A non-in-our-list peer has sent us a message. WTH ?");
 347     //Update the interested state of the peer.
 348     remote_peer->interested = 1;
 349     break;
 350   case MESSAGE_NOTINTERESTED:
 351     XBT_DEBUG("Received a NOTINTERESTED message from %s (%s)", message->mailbox,
 352               message->issuer_host_name);
 353     xbt_assert((remote_peer != NULL),
 354                "A non-in-our-list peer has sent us a message. WTH ?");
 355     remote_peer->interested = 0;
 356     break;
 357   case MESSAGE_UNCHOKE:
 358     xbt_assert((remote_peer != NULL),
 359                "A non-in-our-list peer has sent us a message. WTH ?");
 360     XBT_DEBUG("Received a UNCHOKE message from %s (%s)", message->mailbox,
 361               message->issuer_host_name);
 362     remote_peer->choked_download = 0;
 363     xbt_dict_set_ext(peer->active_peers, (char *) &message->peer_id,
 364                      sizeof(int), remote_peer, NULL);
 365     //Send requests to the peer, since it has unchoked us
 366     send_requests_to_peer(peer, remote_peer);
 367     break;
 368   case MESSAGE_CHOKE:
 369     xbt_assert((remote_peer != NULL),
 370                "A non-in-our-list peer has sent us a message. WTH ?");
 371     XBT_DEBUG("Received a CHOKE message from %s (%s)", message->mailbox,
 372               message->issuer_host_name);
 373     remote_peer->choked_download = 1;
 374     xbt_ex_t e;
 375     TRY {
 376       xbt_dict_remove_ext(peer->active_peers, (char *) &message->peer_id,
 377                           sizeof(int));
 378     }
 379     CATCH(e) {
 380       xbt_ex_free(e);
 381     }
 382     break;
 383   case MESSAGE_HAVE:
 384     XBT_DEBUG("Received a HAVE message from %s (%s) of piece %d",
 385               message->mailbox, message->issuer_host_name, message->index);
 386     xbt_assert((message->index >= 0
 387                 && message->index < FILE_PIECES),
 388                "Wrong HAVE message received");
 389     if (remote_peer->bitfield == NULL)
 390       return;
 391     remote_peer->bitfield[message->index] = '1';
 392     peer->pieces_count[message->index]++;
 393     //If the piece is in our pieces, we tell the peer that we are interested.
 394     if (!remote_peer->am_interested && in_current_pieces(peer, message->index)) {
 395       remote_peer->am_interested = 1;
 396       send_interested(peer, remote_peer->mailbox);
 397     }
 398     break;
 399   case MESSAGE_REQUEST:
 400     xbt_assert((message->index >= 0
 401                 && message->index < FILE_PIECES), "Wrong request received");
 402     if (!remote_peer->choked_upload) {
 403       XBT_DEBUG("Received a REQUEST from %s (%s) for %d (%d,%d)",
 404                 message->mailbox, message->issuer_host_name, message->peer_id,
 405                 message->block_index,
 406                 message->block_index + message->block_length);
 407       if (peer->bitfield[message->index] == '1') {
 408         send_piece(peer, message->mailbox, message->index, 0,
 409                    message->block_index, message->block_length);
 410       }
 411     } else {
 412       XBT_DEBUG("Received a REQUEST from %s (%s) for %d but he is choked.",
 413                 message->mailbox, message->issuer_host_name, message->peer_id);
 414     }
 415     break;
 416   case MESSAGE_PIECE:
 417     xbt_assert((message->index >= 0
 418                 && message->index < FILE_PIECES), "Wrong piece received");
 419     //TODO: Execute à computation.
 420     if (message->stalled) {
 421       XBT_DEBUG("The received piece %d from %s (%s) is STALLED", message->index,
 422                 message->mailbox, message->issuer_host_name);
 423     } else {
 424       XBT_DEBUG("Received piece %d (%d,%d) from %s (%s)", message->index,
 425                 message->block_index,
 426                 message->block_index + message->block_length, message->mailbox,
 427                 message->issuer_host_name);
 428       if (peer->bitfield[message->index] == '0') {
 429         update_bitfield_blocks(peer, message->index, message->block_index,
 430                                message->block_length);
 431         if (piece_complete(peer, message->index)) {
 432           peer->pieces_requested--;
 433           //Removing the piece from our piece list
 434           unsigned i;
 435           int piece_index = -1, piece;
 436           xbt_dynar_foreach(peer->current_pieces, i, piece) {
 437             if (piece == message->index) {
 438               piece_index = i;
 439               break;
 440             }
 441           }
 442           xbt_assert(piece_index != -1, "Received an incorrect piece");
 443           xbt_dynar_remove_at(peer->current_pieces, piece_index, NULL);
 444           //Setting the fact that we have the piece
 445           peer->bitfield[message->index] = '1';
 446           peer->pieces++;
 447           XBT_DEBUG("My status is now %s", peer->bitfield);
 448           //Sending the information to all the peers we are connected to
 449           send_have(peer, message->index);
 450           //sending UNINTERSTED to peers that doesn't have what we want.
 451           update_interested_after_receive(peer);
 452         }
 453       } else {
 454         XBT_DEBUG("However, we already have it");
 455       }
 456     }
 457     break;
 458   case MESSAGE_CANCEL:
 459     break;
 460   }
 461   //Update the peer speed.
 462   if (remote_peer) {
 463     connection_add_speed_value(remote_peer,
 464                                1.0 / (MSG_get_clock() -
 465                                       peer->begin_receive_time));
 466   }
 467   peer->begin_receive_time = MSG_get_clock();
 468
 469   task_message_free(task);
 470 }
 471
 472 /**
 473  * Wait for the node to receive interesting bitfield messages (ie: non empty)
 474  * to be received
 475  * @param deadline peer deadline
 476  * @param peer peer data
 477  */
 478 void wait_for_pieces(peer_t peer, double deadline)
 479 {
 480   int finished = 0;
 481   while (MSG_get_clock() < deadline && !finished) {
 482     if (peer->comm_received == NULL) {
 483       peer->task_received = NULL;
 484       peer->comm_received = MSG_task_irecv(&peer->task_received, peer->mailbox);
 485     }
 486     msg_error_t status = MSG_comm_wait(peer->comm_received, TIMEOUT_MESSAGE);
 487     //free the comm already, we don't need it anymore
 488     MSG_comm_destroy(peer->comm_received);
 489     peer->comm_received = NULL;
 490     if (status == MSG_OK) {
 491       MSG_task_get_data(peer->task_received);
 492       handle_message(peer, peer->task_received);
 493       if (peer->current_piece != -1) {
 494         finished = 1;
 495       }
 496     }
 497   }
 498 }
 499
 500 /**
 501  * Updates the list of who has a piece from a bitfield
 502  * @param peer peer we want to update the list
 503  * @param bitfield bitfield
 504  */
 505 void update_pieces_count_from_bitfield(peer_t peer, char *bitfield)
 506 {
 507   int i;
 508   for (i = 0; i < FILE_PIECES; i++) {
 509     if (bitfield[i] == '1') {
 510       peer->pieces_count[i]++;
 511     }
 512   }
 513 }
 514
 515 /**
 516  * Update the piece the peer is currently interested in.
 517  * There is two cases (as described in "Bittorrent Architecture Protocol", Ryan Toole :
 518  * If the peer has less than 3 pieces, he chooses a piece at random.
 519  * If the peer has more than pieces, he downloads the pieces that are the less
 520  * replicated
 521  */
 522 void update_current_piece(peer_t peer)
 523 {
 524   if (xbt_dynar_length(peer->current_pieces) >= (FILE_PIECES - peer->pieces)) {
 525     return;
 526   }
 527   if (1 || peer->pieces < 3) {
 528     int i = 0;
 529     do {
 530       peer->current_piece =
 531           RngStream_RandInt(peer->stream, 0, FILE_PIECES - 1);;
 532       i++;
 533     } while (!
 534              (peer->bitfield[peer->current_piece] == '0'
 535               && !in_current_pieces(peer, peer->current_piece)));
 536   } else {
 537     //Trivial min algorithm.
 538     int i, min_id = -1;
 539     short min = -1;
 540     for (i = 0; i < FILE_PIECES; i++) {
 541       if (peer->bitfield[i] == '0') {
 542         min = peer->pieces_count[i];
 543         min_id = i;
 544         break;
 545       }
 546     }
 547     xbt_assert((min > -1), "Couldn't find a minimum");
 548     for (i = 1; i < FILE_PIECES; i++) {
 549       if (peer->pieces_count[i] < min && peer->bitfield[i] == '0') {
 550         min = peer->pieces_count[i];
 551         min_id = i;
 552       }
 553     }
 554     peer->current_piece = min_id;
 555   }
 556   xbt_dynar_push_as(peer->current_pieces, int, peer->current_piece);
 557   XBT_DEBUG("New interested piece: %d", peer->current_piece);
 558   xbt_assert((peer->current_piece >= 0 && peer->current_piece < FILE_PIECES),
 559              "Peer want to retrieve a piece that doesn't exist.");
 560 }
 561
 562 /**
 563  * Update the list of current choked and unchoked peers, using the
 564  * choke algorithm
 565  * @param peer the current peer
 566  */
 567 void update_choked_peers(peer_t peer)
 568 {
 569   //update the current round
 570   peer->round = (peer->round + 1) % 3;
 571   char *key;
 572   connection_t peer_choosed = NULL;
 573   //remove a peer from the list
 574   xbt_dict_cursor_t cursor = NULL;
 575   xbt_dict_cursor_first(peer->active_peers, &cursor);
 576   if (xbt_dict_length(peer->active_peers) > 0) {
 577     key = xbt_dict_cursor_get_key(cursor);
 578     connection_t peer_choked = xbt_dict_cursor_get_data(cursor);
 579     if (peer_choked) {
 580       send_choked(peer, peer_choked->mailbox);
 581       peer_choked->choked_upload = 1;
 582     }
 583     xbt_dict_remove_ext(peer->active_peers, key, sizeof(int));
 584   }
 585   xbt_dict_cursor_free(&cursor);
 586
 587         /**
 588          * If we are currently seeding, we unchoke the peer which has
 589          * been unchoke the least time.
 590          */
 591   if (peer->pieces == FILE_PIECES) {
 592     connection_t connection;
 593     double unchoke_time = MSG_get_clock() + 1;
 594
 595     xbt_dict_foreach(peer->peers, cursor, key, connection) {
 596       if (connection->last_unchoke < unchoke_time && connection->interested) {
 597         unchoke_time = connection->last_unchoke;
 598         peer_choosed = connection;
 599       }
 600     }
 601   } else {
 602     //Random optimistic unchoking
 603     if (peer->round == 0) {
 604       int j = 0;
 605       do {
 606         //We choose a random peer to unchoke.
 607         int id_chosen =
 608             RngStream_RandInt(peer->stream, 0,
 609                               xbt_dict_length(peer->peers) - 1);
 610         int i = 0;
 611         connection_t connection;
 612         xbt_dict_foreach(peer->peers, cursor, key, connection) {
 613           if (i == id_chosen) {
 614             peer_choosed = connection;
 615             break;
 616           }
 617           i++;
 618         }
 619         xbt_dict_cursor_free(&cursor);
 620         if (peer_choosed->interested == 0) {
 621           peer_choosed = NULL;
 622         }
 623         j++;
 624       } while (peer_choosed == NULL && j < MAXIMUM_PAIRS);
 625     } else {
 626       //Use the "fastest download" policy.
 627       connection_t connection;
 628       double fastest_speed = 0.0;
 629       xbt_dict_foreach(peer->peers, cursor, key, connection) {
 630         if (connection->peer_speed > fastest_speed && connection->choked_upload
 631             && connection->interested) {
 632           peer_choosed = connection;
 633           fastest_speed = connection->peer_speed;
 634         }
 635       }
 636     }
 637
 638   }
 639   if (peer_choosed != NULL) {
 640     peer_choosed->choked_upload = 0;
 641     xbt_dict_set_ext(peer->active_peers, (char *) &peer_choosed->id,
 642                      sizeof(int), peer_choosed, NULL);
 643     peer_choosed->last_unchoke = MSG_get_clock();
 644     send_unchoked(peer, peer_choosed->mailbox);
 645   }
 646
 647 }
 648
 649 /**
 650  * Updates our "interested" state about peers: send "not interested" to peers
 651  * that don't have any more pieces we want.
 652  * @param peer our peer data
 653  */
 654 void update_interested_after_receive(peer_t peer)
 655 {
 656   char *key;
 657   xbt_dict_cursor_t cursor;
 658   connection_t connection;
 659   unsigned cpt;
 660   int interested, piece;
 661   xbt_dict_foreach(peer->peers, cursor, key, connection) {
 662     interested = 0;
 663     if (connection->am_interested) {
 664       //Check if the peer still has a piece we want.
 665       xbt_dynar_foreach(peer->current_pieces, cpt, piece) {
 666         xbt_assert((piece >= 0), "Wrong piece.");
 667         if (connection->bitfield && connection->bitfield[piece] == '1') {
 668           interested = 1;
 669           break;
 670         }
 671       }
 672       if (!interested) {
 673         connection->am_interested = 0;
 674         send_notinterested(peer, connection->mailbox);
 675       }
 676     }
 677   }
 678 }
 679
 680 void update_bitfield_blocks(peer_t peer, int index, int block_index,
 681                             int block_length)
 682 {
 683   int i;
 684   xbt_assert((index >= 0 && index <= FILE_PIECES), "Wrong piece.");
 685   xbt_assert((block_index >= 0
 686               && block_index <= PIECES_BLOCKS), "Wrong block : %d.",
 687              block_index);
 688   for (i = block_index; i < (block_index + block_length); i++) {
 689     peer->bitfield_blocks[index * PIECES_BLOCKS + i] = '1';
 690   }
 691 }
 692
 693 /**
 694  * Returns if a peer has completed the download of a piece
 695  */
 696 int piece_complete(peer_t peer, int index)
 697 {
 698   int i;
 699   for (i = 0; i < PIECES_BLOCKS; i++) {
 700     if (peer->bitfield_blocks[index * PIECES_BLOCKS + i] == '0') {
 701       return 0;
 702     }
 703   }
 704   return 1;
 705 }
 706
 707 /**
 708  * Returns the first block that a peer doesn't have in a piece
 709  */
 710 int get_first_block(peer_t peer, int piece)
 711 {
 712   int i;
 713   for (i = 0; i < PIECES_BLOCKS; i++) {
 714     if (peer->bitfield_blocks[piece * PIECES_BLOCKS + i] == '0') {
 715       return i;
 716     }
 717   }
 718   return -1;
 719 }
 720
 721 /**
 722  * Send request messages to a peer that have unchoked us
 723  * @param peer peer
 724  * @param remote_peer peer data to the peer we want to send the request
 725  */
 726 void send_requests_to_peer(peer_t peer, connection_t remote_peer)
 727 {
 728   unsigned i;
 729   int piece, block_index, block_length;
 730   xbt_dynar_foreach(peer->current_pieces, i, piece) {
 731     if (remote_peer->bitfield && remote_peer->bitfield[piece] == '1') {
 732       block_index = get_first_block(peer, piece);
 733       if (block_index != -1) {
 734         block_length = PIECES_BLOCKS - block_index;
 735         block_length = min(BLOCKS_REQUESTED, block_length);
 736         send_request(peer, remote_peer->mailbox, piece, block_index,
 737                      block_length);
 738         break;
 739       }
 740     }
 741   }
 742 }
 743
 744 /**
 745  * Find the peers that have the current interested piece and send them
 746  * the "interested" message
 747  */
 748 void send_interested_to_peers(peer_t peer)
 749 {
 750   char *key;
 751   xbt_dict_cursor_t cursor = NULL;
 752   connection_t connection;
 753   xbt_assert((peer->current_piece != -1),
 754              "Tried to send a interested message wheras the current_piece is -1");
 755   xbt_dict_foreach(peer->peers, cursor, key, connection) {
 756     if (connection->bitfield
 757         && connection->bitfield[peer->current_piece] == '1') {
 758       connection->am_interested = 1;
 759       msg_task_t task =
 760           task_message_new(MESSAGE_INTERESTED, peer->hostname, peer->mailbox,
 761                            peer->id, task_message_size(MESSAGE_INTERESTED));
 762       MSG_task_dsend(task, connection->mailbox, task_message_free);
 763       XBT_DEBUG("Send INTERESTED to %s", connection->mailbox);
 764     }
 765   }
 766   peer->current_piece = -1;
 767   peer->pieces_requested++;
 768 }
 769
 770 /**
 771  * Send a "interested" message to a peer
 772  * @param peer peer data
 773  * @param mailbox destination mailbox
 774  */
 775 void send_interested(peer_t peer, const char *mailbox)
 776 {
 777   msg_task_t task =
 778       task_message_new(MESSAGE_INTERESTED, peer->hostname, peer->mailbox,
 779                        peer->id, task_message_size(MESSAGE_INTERESTED));
 780   MSG_task_dsend(task, mailbox, task_message_free);
 781   XBT_DEBUG("Sending INTERESTED to %s", mailbox);
 782
 783 }
 784
 785 /**
 786  * Send a "not interested" message to a peer
 787  * @param peer peer data
 788  * @param mailbox destination mailbox
 789  */
 790 void send_notinterested(peer_t peer, const char *mailbox)
 791 {
 792   msg_task_t task =
 793       task_message_new(MESSAGE_NOTINTERESTED, peer->hostname, peer->mailbox,
 794                        peer->id, task_message_size(MESSAGE_NOTINTERESTED));
 795   MSG_task_dsend(task, mailbox, task_message_free);
 796   XBT_DEBUG("Sending NOTINTERESTED to %s", mailbox);
 797
 798 }
 799
 800 /**
 801  * Send a handshake message to all the peers the peer has.
 802  * @param peer peer data
 803  */
 804 void send_handshake_all(peer_t peer)
 805 {
 806   connection_t remote_peer;
 807   xbt_dict_cursor_t cursor = NULL;
 808   char *key;
 809   xbt_dict_foreach(peer->peers, cursor, key, remote_peer) {
 810     msg_task_t task =
 811         task_message_new(MESSAGE_HANDSHAKE, peer->hostname, peer->mailbox,
 812                          peer->id, task_message_size(MESSAGE_HANDSHAKE));
 813     MSG_task_dsend(task, remote_peer->mailbox, task_message_free);
 814     XBT_DEBUG("Sending a HANDSHAKE to %s", remote_peer->mailbox);
 815   }
 816 }
 817
 818 /**
 819  * Send a "handshake" message to an user
 820  * @param peer peer data
 821  * @param mailbox mailbox where to we send the message
 822  */
 823 void send_handshake(peer_t peer, const char *mailbox)
 824 {
 825   msg_task_t task =
 826       task_message_new(MESSAGE_HANDSHAKE, peer->hostname, peer->mailbox,
 827                        peer->id, task_message_size(MESSAGE_HANDSHAKE));
 828   MSG_task_dsend(task, mailbox, task_message_free);
 829   XBT_DEBUG("Sending a HANDSHAKE to %s", mailbox);
 830 }
 831
 832 /**
 833  * Send a "choked" message to a peer.
 834  */
 835 void send_choked(peer_t peer, const char *mailbox)
 836 {
 837   XBT_DEBUG("Sending a CHOKE to %s", mailbox);
 838   msg_task_t task =
 839       task_message_new(MESSAGE_CHOKE, peer->hostname, peer->mailbox,
 840                        peer->id, task_message_size(MESSAGE_CHOKE));
 841   MSG_task_dsend(task, mailbox, task_message_free);
 842 }
 843
 844 /**
 845  * Send a "unchoked" message to a peer
 846  */
 847 void send_unchoked(peer_t peer, const char *mailbox)
 848 {
 849   XBT_DEBUG("Sending a UNCHOKE to %s", mailbox);
 850   msg_task_t task =
 851       task_message_new(MESSAGE_UNCHOKE, peer->hostname, peer->mailbox,
 852                        peer->id, task_message_size(MESSAGE_UNCHOKE));
 853   MSG_task_dsend(task, mailbox, task_message_free);
 854 }
 855
 856 /**
 857  * Send a "HAVE" message to all peers we are connected to
 858  */
 859 void send_have(peer_t peer, int piece)
 860 {
 861   XBT_DEBUG("Sending HAVE message to all my peers");
 862   connection_t remote_peer;
 863   xbt_dict_cursor_t cursor = NULL;
 864   char *key;
 865   xbt_dict_foreach(peer->peers, cursor, key, remote_peer) {
 866     msg_task_t task =
 867         task_message_index_new(MESSAGE_HAVE, peer->hostname, peer->mailbox,
 868                                peer->id, piece, task_message_size(MESSAGE_HAVE));
 869     MSG_task_dsend(task, remote_peer->mailbox, task_message_free);
 870   }
 871 }
 872
 873 /**
 874  * Send a bitfield message to all the peers the peer has.
 875  * @param peer peer data
 876  */
 877 void send_bitfield(peer_t peer, const char *mailbox)
 878 {
 879   XBT_DEBUG("Sending a BITFIELD to %s", mailbox);
 880   msg_task_t task =
 881       task_message_bitfield_new(peer->hostname, peer->mailbox, peer->id,
 882                                 peer->bitfield, FILE_PIECES);
 883   //Async send and append to pending sends
 884   msg_comm_t comm = MSG_task_isend(task, mailbox);
 885   xbt_dynar_push(peer->pending_sends, &comm);
 886 }
 887
 888 /**
 889  * Send a "request" message to a pair, containing a request for a piece
 890  */
 891 void send_request(peer_t peer, const char *mailbox, int piece, int block_index,
 892                   int block_length)
 893 {
 894   XBT_DEBUG("Sending a REQUEST to %s for piece %d (%d,%d)", mailbox, piece,
 895             block_index, block_length);
 896   msg_task_t task =
 897       task_message_request_new(peer->hostname, peer->mailbox, peer->id, piece,
 898                                block_index, block_length);
 899   MSG_task_dsend(task, mailbox, task_message_free);
 900 }
 901
 902 /**
 903  * Send a "piece" message to a pair, containing a piece of the file
 904  */
 905 void send_piece(peer_t peer, const char *mailbox, int piece, int stalled,
 906                 int block_index, int block_length)
 907 {
 908   XBT_DEBUG("Sending the PIECE %d (%d,%d) to %s", piece, block_index,
 909             block_length, mailbox);
 910   xbt_assert(piece >= 0, "Tried to send a piece that doesn't exist.");
 911   xbt_assert((peer->bitfield[piece] == '1'),
 912              "Tried to send a piece that we doesn't have.");
 913   msg_task_t task =
 914       task_message_piece_new(peer->hostname, peer->mailbox, peer->id, piece,
 915                              stalled, block_index, block_length, BLOCK_SIZE);
 916   MSG_task_dsend(task, mailbox, task_message_free);
 917 }
 918
 919 int in_current_pieces(peer_t peer, int piece)
 920 {
 921   unsigned i;
 922   int is_in = 0, peer_piece;
 923   xbt_dynar_foreach(peer->current_pieces, i, peer_piece) {
 924     if (peer_piece == piece) {
 925       is_in = 1;
 926       break;
 927     }
 928   }
 929   return is_in;
 930 }