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