examples/msg/app-bittorrent/peer.c

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