examples/msg/bittorrent/peer.c

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