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   }
 432   //Update the peer speed.
 433   if (remote_peer) {
 434     connection_add_speed_value(remote_peer,
 435                                1.0 / (MSG_get_clock() -
 436                                       peer->begin_receive_time));
 437   }
 438   peer->begin_receive_time = MSG_get_clock();
 439
 440   task_message_free(task);
 441 }
 442
 443 /**
 444  * Wait for the node to receive interesting bitfield messages (ie: non empty)
 445  * to be received
 446  * @param deadline peer deadline
 447  * @param peer peer data
 448  */
 449 void wait_for_pieces(peer_t peer, double deadline)
 450 {
 451   int finished = 0;
 452   while (MSG_get_clock() < deadline && !finished) {
 453     if (peer->comm_received == NULL) {
 454       peer->task_received = NULL;
 455       peer->comm_received = MSG_task_irecv(&peer->task_received, peer->mailbox);
 456     }
 457     msg_error_t status = MSG_comm_wait(peer->comm_received, TIMEOUT_MESSAGE);
 458     //free the comm already, we don't need it anymore
 459     MSG_comm_destroy(peer->comm_received);
 460     peer->comm_received = NULL;
 461     if (status == MSG_OK) {
 462       MSG_task_get_data(peer->task_received);
 463       handle_message(peer, peer->task_received);
 464       if (peer->current_piece != -1) {
 465         finished = 1;
 466       }
 467     }
 468   }
 469 }
 470
 471 /**
 472  * Updates the list of who has a piece from a bitfield
 473  * @param peer peer we want to update the list
 474  * @param bitfield bitfield
 475  */
 476 void update_pieces_count_from_bitfield(peer_t peer, char *bitfield)
 477 {
 478   int i;
 479   for (i = 0; i < FILE_PIECES; i++) {
 480     if (bitfield[i] == '1') {
 481       peer->pieces_count[i]++;
 482     }
 483   }
 484 }
 485
 486 /**
 487  * Update the piece the peer is currently interested in.
 488  * There is two cases (as described in "Bittorrent Architecture Protocol", Ryan Toole :
 489  * If the peer has less than 3 pieces, he chooses a piece at random.
 490  * If the peer has more than pieces, he downloads the pieces that are the less
 491  * replicated
 492  */
 493 void update_current_piece(peer_t peer)
 494 {
 495   if (xbt_dynar_length(peer->current_pieces) >= (FILE_PIECES - peer->pieces)) {
 496     return;
 497   }
 498   if (1 || peer->pieces < 3) {
 499     int i = 0;
 500     do {
 501       peer->current_piece =
 502           RngStream_RandInt(peer->stream, 0, FILE_PIECES - 1);;
 503       i++;
 504     } while (!
 505              (peer->bitfield[peer->current_piece] == '0'
 506               && !in_current_pieces(peer, peer->current_piece)));
 507   } else {
 508     //Trivial min algorithm.
 509     int i, min_id = -1;
 510     short min = -1;
 511     for (i = 0; i < FILE_PIECES; i++) {
 512       if (peer->bitfield[i] == '0') {
 513         min = peer->pieces_count[i];
 514         min_id = i;
 515         break;
 516       }
 517     }
 518     xbt_assert((min > -1), "Couldn't find a minimum");
 519     for (i = 1; i < FILE_PIECES; i++) {
 520       if (peer->pieces_count[i] < min && peer->bitfield[i] == '0') {
 521         min = peer->pieces_count[i];
 522         min_id = i;
 523       }
 524     }
 525     peer->current_piece = min_id;
 526   }
 527   xbt_dynar_push_as(peer->current_pieces, int, peer->current_piece);
 528   XBT_DEBUG("New interested piece: %d", peer->current_piece);
 529   xbt_assert((peer->current_piece >= 0 && peer->current_piece < FILE_PIECES),
 530              "Peer want to retrieve a piece that doesn't exist.");
 531 }
 532
 533 /**
 534  * Update the list of current choked and unchoked peers, using the
 535  * choke algorithm
 536  * @param peer the current peer
 537  */
 538 void update_choked_peers(peer_t peer)
 539 {
 540   //update the current round
 541   peer->round = (peer->round + 1) % 3;
 542   char *key;
 543   connection_t peer_choosed = NULL;
 544   //remove a peer from the list
 545   xbt_dict_cursor_t cursor = NULL;
 546   xbt_dict_cursor_first(peer->active_peers, &cursor);
 547   if (xbt_dict_length(peer->active_peers) > 0) {
 548     key = xbt_dict_cursor_get_key(cursor);
 549     connection_t peer_choked = xbt_dict_cursor_get_data(cursor);
 550     if (peer_choked) {
 551       send_choked(peer, peer_choked->mailbox);
 552       peer_choked->choked_upload = 1;
 553     }
 554     xbt_dict_remove_ext(peer->active_peers, key, sizeof(int));
 555   }
 556   xbt_dict_cursor_free(&cursor);
 557
 558         /**
 559          * If we are currently seeding, we unchoke the peer which has
 560          * been unchoke the least time.
 561          */
 562   if (peer->pieces == FILE_PIECES) {
 563     connection_t connection;
 564     double unchoke_time = MSG_get_clock() + 1;
 565
 566     xbt_dict_foreach(peer->peers, cursor, key, connection) {
 567       if (connection->last_unchoke < unchoke_time && connection->interested) {
 568         unchoke_time = connection->last_unchoke;
 569         peer_choosed = connection;
 570       }
 571     }
 572   } else {
 573     //Random optimistic unchoking
 574     if (peer->round == 0) {
 575       int j = 0;
 576       do {
 577         //We choose a random peer to unchoke.
 578         int id_chosen =
 579             RngStream_RandInt(peer->stream, 0,
 580                               xbt_dict_length(peer->peers) - 1);
 581         int i = 0;
 582         connection_t connection;
 583         xbt_dict_foreach(peer->peers, cursor, key, connection) {
 584           if (i == id_chosen) {
 585             peer_choosed = connection;
 586             break;
 587           }
 588           i++;
 589         }
 590         xbt_dict_cursor_free(&cursor);
 591         if (peer_choosed->interested == 0) {
 592           peer_choosed = NULL;
 593         }
 594         j++;
 595       } while (peer_choosed == NULL && j < MAXIMUM_PAIRS);
 596     } else {
 597       //Use the "fastest download" policy.
 598       connection_t connection;
 599       double fastest_speed = 0.0;
 600       xbt_dict_foreach(peer->peers, cursor, key, connection) {
 601         if (connection->peer_speed > fastest_speed && connection->choked_upload
 602             && connection->interested) {
 603           peer_choosed = connection;
 604           fastest_speed = connection->peer_speed;
 605         }
 606       }
 607     }
 608
 609   }
 610   if (peer_choosed != NULL) {
 611     peer_choosed->choked_upload = 0;
 612     xbt_dict_set_ext(peer->active_peers, (char *) &peer_choosed->id,
 613                      sizeof(int), peer_choosed, NULL);
 614     peer_choosed->last_unchoke = MSG_get_clock();
 615     send_unchoked(peer, peer_choosed->mailbox);
 616   }
 617
 618 }
 619
 620 /**
 621  * Updates our "interested" state about peers: send "not interested" to peers
 622  * that don't have any more pieces we want.
 623  * @param peer our peer data
 624  */
 625 void update_interested_after_receive(peer_t peer)
 626 {
 627   char *key;
 628   xbt_dict_cursor_t cursor;
 629   connection_t connection;
 630   unsigned cpt;
 631   int interested, piece;
 632   xbt_dict_foreach(peer->peers, cursor, key, connection) {
 633     interested = 0;
 634     if (connection->am_interested) {
 635       //Check if the peer still has a piece we want.
 636       xbt_dynar_foreach(peer->current_pieces, cpt, piece) {
 637         xbt_assert((piece >= 0), "Wrong piece.");
 638         if (connection->bitfield && connection->bitfield[piece] == '1') {
 639           interested = 1;
 640           break;
 641         }
 642       }
 643       if (!interested) {
 644         connection->am_interested = 0;
 645         send_notinterested(peer, connection->mailbox);
 646       }
 647     }
 648   }
 649 }
 650
 651 void update_bitfield_blocks(peer_t peer, int index, int block_index,
 652                             int block_length)
 653 {
 654   int i;
 655   xbt_assert((index >= 0 && index <= FILE_PIECES), "Wrong piece.");
 656   xbt_assert((block_index >= 0
 657               && block_index <= PIECES_BLOCKS), "Wrong block : %d.",
 658              block_index);
 659   for (i = block_index; i < (block_index + block_length); i++) {
 660     peer->bitfield_blocks[index * PIECES_BLOCKS + i] = '1';
 661   }
 662 }
 663
 664 /**
 665  * Returns if a peer has completed the download of a piece
 666  */
 667 int piece_complete(peer_t peer, int index)
 668 {
 669   int i;
 670   for (i = 0; i < PIECES_BLOCKS; i++) {
 671     if (peer->bitfield_blocks[index * PIECES_BLOCKS + i] == '0') {
 672       return 0;
 673     }
 674   }
 675   return 1;
 676 }
 677
 678 /**
 679  * Returns the first block that a peer doesn't have in a piece
 680  */
 681 int get_first_block(peer_t peer, int piece)
 682 {
 683   int i;
 684   for (i = 0; i < PIECES_BLOCKS; i++) {
 685     if (peer->bitfield_blocks[piece * PIECES_BLOCKS + i] == '0') {
 686       return i;
 687     }
 688   }
 689   return -1;
 690 }
 691
 692 /**
 693  * Send request messages to a peer that have unchoked us
 694  * @param peer peer
 695  * @param remote_peer peer data to the peer we want to send the request
 696  */
 697 void send_requests_to_peer(peer_t peer, connection_t remote_peer)
 698 {
 699   unsigned i;
 700   int piece, block_index, block_length;
 701   xbt_dynar_foreach(peer->current_pieces, i, piece) {
 702     if (remote_peer->bitfield && remote_peer->bitfield[piece] == '1') {
 703       block_index = get_first_block(peer, piece);
 704       if (block_index != -1) {
 705         block_length = PIECES_BLOCKS - block_index;
 706         block_length = min(BLOCKS_REQUESTED, block_length);
 707         send_request(peer, remote_peer->mailbox, piece, block_index,
 708                      block_length);
 709         break;
 710       }
 711     }
 712   }
 713 }
 714
 715 /**
 716  * Find the peers that have the current interested piece and send them
 717  * the "interested" message
 718  */
 719 void send_interested_to_peers(peer_t peer)
 720 {
 721   char *key;
 722   xbt_dict_cursor_t cursor = NULL;
 723   connection_t connection;
 724   xbt_assert((peer->current_piece != -1),
 725              "Tried to send a interested message wheras the current_piece is -1");
 726   xbt_dict_foreach(peer->peers, cursor, key, connection) {
 727     if (connection->bitfield
 728         && connection->bitfield[peer->current_piece] == '1') {
 729       connection->am_interested = 1;
 730       msg_task_t task =
 731           task_message_new(MESSAGE_INTERESTED, peer->hostname, peer->mailbox,
 732                            peer->id, task_message_size(MESSAGE_INTERESTED));
 733       MSG_task_dsend(task, connection->mailbox, task_message_free);
 734       XBT_DEBUG("Send INTERESTED to %s", connection->mailbox);
 735     }
 736   }
 737   peer->current_piece = -1;
 738   peer->pieces_requested++;
 739 }
 740
 741 /**
 742  * Send a "interested" message to a peer
 743  * @param peer peer data
 744  * @param mailbox destination mailbox
 745  */
 746 void send_interested(peer_t peer, const char *mailbox)
 747 {
 748   msg_task_t task =
 749       task_message_new(MESSAGE_INTERESTED, peer->hostname, peer->mailbox,
 750                        peer->id, task_message_size(MESSAGE_INTERESTED));
 751   MSG_task_dsend(task, mailbox, task_message_free);
 752   XBT_DEBUG("Sending INTERESTED to %s", mailbox);
 753
 754 }
 755
 756 /**
 757  * Send a "not interested" message to a peer
 758  * @param peer peer data
 759  * @param mailbox destination mailbox
 760  */
 761 void send_notinterested(peer_t peer, const char *mailbox)
 762 {
 763   msg_task_t task =
 764       task_message_new(MESSAGE_NOTINTERESTED, peer->hostname, peer->mailbox,
 765                        peer->id, task_message_size(MESSAGE_NOTINTERESTED));
 766   MSG_task_dsend(task, mailbox, task_message_free);
 767   XBT_DEBUG("Sending NOTINTERESTED to %s", mailbox);
 768
 769 }
 770
 771 /**
 772  * Send a handshake message to all the peers the peer has.
 773  * @param peer peer data
 774  */
 775 void send_handshake_all(peer_t peer)
 776 {
 777   connection_t remote_peer;
 778   xbt_dict_cursor_t cursor = NULL;
 779   char *key;
 780   xbt_dict_foreach(peer->peers, cursor, key, remote_peer) {
 781     msg_task_t task =
 782         task_message_new(MESSAGE_HANDSHAKE, peer->hostname, peer->mailbox,
 783                          peer->id, task_message_size(MESSAGE_HANDSHAKE));
 784     MSG_task_dsend(task, remote_peer->mailbox, task_message_free);
 785     XBT_DEBUG("Sending a HANDSHAKE to %s", remote_peer->mailbox);
 786   }
 787 }
 788
 789 /**
 790  * Send a "handshake" message to an user
 791  * @param peer peer data
 792  * @param mailbox mailbox where to we send the message
 793  */
 794 void send_handshake(peer_t peer, const char *mailbox)
 795 {
 796   msg_task_t task =
 797       task_message_new(MESSAGE_HANDSHAKE, peer->hostname, peer->mailbox,
 798                        peer->id, task_message_size(MESSAGE_HANDSHAKE));
 799   MSG_task_dsend(task, mailbox, task_message_free);
 800   XBT_DEBUG("Sending a HANDSHAKE to %s", mailbox);
 801 }
 802
 803 /**
 804  * Send a "choked" message to a peer.
 805  */
 806 void send_choked(peer_t peer, const char *mailbox)
 807 {
 808   XBT_DEBUG("Sending a CHOKE to %s", mailbox);
 809   msg_task_t task =
 810       task_message_new(MESSAGE_CHOKE, peer->hostname, peer->mailbox,
 811                        peer->id, task_message_size(MESSAGE_CHOKE));
 812   MSG_task_dsend(task, mailbox, task_message_free);
 813 }
 814
 815 /**
 816  * Send a "unchoked" message to a peer
 817  */
 818 void send_unchoked(peer_t peer, const char *mailbox)
 819 {
 820   XBT_DEBUG("Sending a UNCHOKE to %s", mailbox);
 821   msg_task_t task =
 822       task_message_new(MESSAGE_UNCHOKE, peer->hostname, peer->mailbox,
 823                        peer->id, task_message_size(MESSAGE_UNCHOKE));
 824   MSG_task_dsend(task, mailbox, task_message_free);
 825 }
 826
 827 /**
 828  * Send a "HAVE" message to all peers we are connected to
 829  */
 830 void send_have(peer_t peer, int piece)
 831 {
 832   XBT_DEBUG("Sending HAVE message to all my peers");
 833   connection_t remote_peer;
 834   xbt_dict_cursor_t cursor = NULL;
 835   char *key;
 836   xbt_dict_foreach(peer->peers, cursor, key, remote_peer) {
 837     msg_task_t task =
 838         task_message_index_new(MESSAGE_HAVE, peer->hostname, peer->mailbox,
 839                                peer->id, piece, task_message_size(MESSAGE_HAVE));
 840     MSG_task_dsend(task, remote_peer->mailbox, task_message_free);
 841   }
 842 }
 843
 844 /**
 845  * Send a bitfield message to all the peers the peer has.
 846  * @param peer peer data
 847  */
 848 void send_bitfield(peer_t peer, const char *mailbox)
 849 {
 850   XBT_DEBUG("Sending a BITFIELD to %s", mailbox);
 851   msg_task_t task =
 852       task_message_bitfield_new(peer->hostname, peer->mailbox, peer->id,
 853                                 peer->bitfield, FILE_PIECES);
 854   MSG_task_dsend(task, mailbox, task_message_free);
 855 }
 856
 857 /**
 858  * Send a "request" message to a pair, containing a request for a piece
 859  */
 860 void send_request(peer_t peer, const char *mailbox, int piece, int block_index,
 861                   int block_length)
 862 {
 863   XBT_DEBUG("Sending a REQUEST to %s for piece %d (%d,%d)", mailbox, piece,
 864             block_index, block_length);
 865   msg_task_t task =
 866       task_message_request_new(peer->hostname, peer->mailbox, peer->id, piece,
 867                                block_index, block_length);
 868   MSG_task_dsend(task, mailbox, task_message_free);
 869 }
 870
 871 /**
 872  * Send a "piece" message to a pair, containing a piece of the file
 873  */
 874 void send_piece(peer_t peer, const char *mailbox, int piece, int stalled,
 875                 int block_index, int block_length)
 876 {
 877   XBT_DEBUG("Sending the PIECE %d (%d,%d) to %s", piece, block_index,
 878             block_length, mailbox);
 879   xbt_assert(piece >= 0, "Tried to send a piece that doesn't exist.");
 880   xbt_assert((peer->bitfield[piece] == '1'),
 881              "Tried to send a piece that we doesn't have.");
 882   msg_task_t task =
 883       task_message_piece_new(peer->hostname, peer->mailbox, peer->id, piece,
 884                              stalled, block_index, block_length, BLOCK_SIZE);
 885   MSG_task_dsend(task, mailbox, task_message_free);
 886 }
 887
 888 int in_current_pieces(peer_t peer, int piece)
 889 {
 890   unsigned i;
 891   int is_in = 0, peer_piece;
 892   xbt_dynar_foreach(peer->current_pieces, i, peer_piece) {
 893     if (peer_piece == piece) {
 894       is_in = 1;
 895       break;
 896     }
 897   }
 898   return is_in;
 899 }