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