Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
-Cleaning a bit the code
[simgrid.git] / examples / msg / bittorrent / peer.c
index 9219ee1..b0ddebc 100644 (file)
@@ -1,5 +1,5 @@
-  /* Copyright (c) 2012. The SimGrid Team.
  * All rights reserved.                                                     */
+/* Copyright (c) 2012. The SimGrid Team.
+ * All rights reserved.                                                     */
 
 /* This program is free software; you can redistribute it and/or modify it
  * under the terms of the license (GNU LGPL) which comes with this package. */
@@ -9,23 +9,36 @@
 #include "messages.h"
 #include <msg/msg.h>
 #include <xbt/RngStream.h>
+#include <limits.h>
 
 XBT_LOG_NEW_DEFAULT_CATEGORY(msg_peers, "Messages specific for the peers");
 
 //TODO: Let users change this
 /*
  * File transfered data
- *
+ * For the test, default values are :
  * File size: 10 pieces * 5 blocks/piece * 16384 bytes/block = 819200 bytes
  */
-static int FILE_SIZE = 10 * 5 * 16384;
-static int FILE_PIECES = 10;
 
-static int PIECES_BLOCKS = 5;
-static int BLOCK_SIZE = 16384;
-static int BLOCKS_REQUESTED = 2;
+#define FILE_PIECES  10
+#define PIECES_BLOCKS 5
+#define BLOCK_SIZE  16384
 
 /**
+ *  Number of blocks asked by each request
+ */
+#define BLOCKS_REQUESTED 2
+
+
+static const int FILE_SIZE = FILE_PIECES * PIECES_BLOCKS * BLOCK_SIZE;
+
+void request_new_piece_to_peer(peer_t peer, connection_t remote_peer);
+void send_request_to_peer(peer_t peer, connection_t remote_peer, int piece);
+void remove_current_piece(peer_t peer, connection_t remote_peer,
+                         int current_piece);
+
+
+  /**
  * Peer main function
  */
 int peer(int argc, char *argv[])
@@ -55,6 +68,7 @@ int peer(int argc, char *argv[])
       seed_loop(&peer, deadline);
     } else {
       leech_loop(&peer, deadline);
+//      XBT_INFO("%d becomes a seeder", peer.id);
       seed_loop(&peer, deadline);
     }
   } else {
@@ -85,15 +99,18 @@ void leech_loop(peer_t peer, double deadline)
    */
   send_handshake_all(peer);
   //Wait for at least one "bitfield" message.
-  wait_for_pieces(peer, deadline);
+//  wait_for_pieces(peer, deadline);
   XBT_DEBUG("Starting main leech loop");
+
   while (MSG_get_clock() < deadline && peer->pieces < FILE_PIECES) {
     if (peer->comm_received == NULL) {
+//      XBT_INFO("irecv");
       peer->task_received = NULL;
       peer->comm_received =
        MSG_task_irecv(&peer->task_received, peer->mailbox);
     }
     if (MSG_comm_test(peer->comm_received)) {
+//      XBT_INFO("comm_test OK");
       msg_error_t status = MSG_comm_get_status(peer->comm_received);
       MSG_comm_destroy(peer->comm_received);
       peer->comm_received = NULL;
@@ -102,14 +119,6 @@ void leech_loop(peer_t peer, double deadline)
       }
     } else {
       handle_pending_sends(peer);
-      if (peer->current_piece != -1) {
-       send_interested_to_peers(peer);
-      } else {
-       //If the current interested pieces is < MAX
-       if (peer->pieces_requested < MAX_PIECES) {
-         update_current_piece(peer);
-       }
-      }
       //We don't execute the choke algorithm if we don't already have a piece
       if (MSG_get_clock() >= next_choked_update && peer->pieces > 0) {
        update_choked_peers(peer);
@@ -237,12 +246,10 @@ void peer_init(peer_t peer, int id, int seed)
   peer->pieces = 0;
 
   peer->pieces_count = xbt_new0(short, FILE_PIECES);
-  peer->pieces_requested = 0;
 
   peer->current_pieces = xbt_dynar_new(sizeof(int), NULL);
-  peer->current_piece = -1;
 
-  peer->stream = MSG_host_get_data(MSG_host_self());
+  peer->stream = RngStream_CreateStream("");
   peer->comm_received = NULL;
 
   peer->round = 0;
@@ -268,6 +275,8 @@ void peer_free(peer_t peer)
   xbt_free(peer->pieces_count);
   xbt_free(peer->bitfield);
   xbt_free(peer->bitfield_blocks);
+
+  RngStream_DeleteStream(&peer->stream);
 }
 
 /**
@@ -280,6 +289,19 @@ int has_finished(char *bitfield)
           NULL) ? 1 : 0);
 }
 
+int nb_interested_peers(peer_t peer)
+{
+  xbt_dict_cursor_t cursor = NULL;
+  char *key;
+  connection_t connection;
+  int nb = 0;
+  xbt_dict_foreach(peer->peers, cursor, key, connection) {
+    if (connection->interested)
+      nb++;
+  }
+  return nb;
+}
+
 /**
  * Handle pending sends and remove those which are done
  * @param peer Peer data
@@ -306,6 +328,27 @@ void handle_pending_sends(peer_t peer)
   }
 }
 
+void update_active_peers_set(peer_t peer, connection_t remote_peer)
+{
+
+  if (remote_peer->interested && !remote_peer->choked_upload) {
+    //add in the active peers set
+    xbt_dict_set_ext(peer->active_peers, (char *) &remote_peer->id,
+                    sizeof(int), remote_peer, NULL);
+  } else {
+    //remove
+    xbt_ex_t e;
+    TRY {
+      xbt_dict_remove_ext(peer->active_peers, (char *) &remote_peer->id,
+                         sizeof(int));
+    }
+    CATCH(e) {
+      xbt_ex_free(e);
+    }
+  }
+}
+
+
 /**
  * Handle a received message sent by another peer
  * @param peer Peer data
@@ -339,9 +382,11 @@ void handle_message(peer_t peer, msg_task_t task)
     update_pieces_count_from_bitfield(peer, message->bitfield);
     //Store the bitfield
     remote_peer->bitfield = xbt_strdup(message->bitfield);
-    //Update the current piece
-    if (peer->current_piece == -1 && peer->pieces < FILE_PIECES) {
-      update_current_piece(peer);
+    xbt_assert(!remote_peer->am_interested,
+              "Should not be interested at first");
+    if (is_interested(peer, remote_peer)) {
+      remote_peer->am_interested = 1;
+      send_interested(peer, message->mailbox);
     }
     break;
   case MESSAGE_INTERESTED:
@@ -351,6 +396,7 @@ void handle_message(peer_t peer, msg_task_t task)
               "A non-in-our-list peer has sent us a message. WTH ?");
     //Update the interested state of the peer.
     remote_peer->interested = 1;
+    update_active_peers_set(peer, remote_peer);
     break;
   case MESSAGE_NOTINTERESTED:
     XBT_DEBUG("Received a NOTINTERESTED message from %s (%s)",
@@ -358,6 +404,7 @@ void handle_message(peer_t peer, msg_task_t task)
     xbt_assert((remote_peer != NULL),
               "A non-in-our-list peer has sent us a message. WTH ?");
     remote_peer->interested = 0;
+    update_active_peers_set(peer, remote_peer);
     break;
   case MESSAGE_UNCHOKE:
     xbt_assert((remote_peer != NULL),
@@ -365,10 +412,9 @@ void handle_message(peer_t peer, msg_task_t task)
     XBT_DEBUG("Received a UNCHOKE message from %s (%s)", message->mailbox,
              message->issuer_host_name);
     remote_peer->choked_download = 0;
-    xbt_dict_set_ext(peer->active_peers, (char *) &message->peer_id,
-                    sizeof(int), remote_peer, NULL);
     //Send requests to the peer, since it has unchoked us
-    send_requests_to_peer(peer, remote_peer);
+    if (remote_peer->am_interested)
+      request_new_piece_to_peer(peer, remote_peer);
     break;
   case MESSAGE_CHOKE:
     xbt_assert((remote_peer != NULL),
@@ -376,30 +422,23 @@ void handle_message(peer_t peer, msg_task_t task)
     XBT_DEBUG("Received a CHOKE message from %s (%s)", message->mailbox,
              message->issuer_host_name);
     remote_peer->choked_download = 1;
-    xbt_ex_t e;
-    TRY {
-      xbt_dict_remove_ext(peer->active_peers, (char *) &message->peer_id,
-                         sizeof(int));
-    }
-    CATCH(e) {
-      xbt_ex_free(e);
-    }
+    remove_current_piece(peer, remote_peer, remote_peer->current_piece);
     break;
   case MESSAGE_HAVE:
     XBT_DEBUG("Received a HAVE message from %s (%s) of piece %d",
              message->mailbox, message->issuer_host_name, message->index);
+    xbt_assert(remote_peer->bitfield, "bitfield not received");
     xbt_assert((message->index >= 0
                && message->index < FILE_PIECES),
               "Wrong HAVE message received");
-    if (remote_peer->bitfield == NULL)
-      return;
     remote_peer->bitfield[message->index] = '1';
     peer->pieces_count[message->index]++;
     //If the piece is in our pieces, we tell the peer that we are interested.
-    if (!remote_peer->am_interested
-       && in_current_pieces(peer, message->index)) {
+    if (!remote_peer->am_interested && peer->bitfield[message->index] == '0') {
       remote_peer->am_interested = 1;
-      send_interested(peer, remote_peer->mailbox);
+      send_interested(peer, message->mailbox);
+      if (!remote_peer->choked_download)
+       request_new_piece_to_peer(peer, remote_peer);
     }
     break;
   case MESSAGE_REQUEST:
@@ -436,18 +475,8 @@ void handle_message(peer_t peer, msg_task_t task)
        update_bitfield_blocks(peer, message->index, message->block_index,
                               message->block_length);
        if (piece_complete(peer, message->index)) {
-         peer->pieces_requested--;
          //Removing the piece from our piece list
-         unsigned i;
-         int piece_index = -1, piece;
-         xbt_dynar_foreach(peer->current_pieces, i, piece) {
-           if (piece == message->index) {
-             piece_index = i;
-             break;
-           }
-         }
-         xbt_assert(piece_index != -1, "Received an incorrect piece");
-         xbt_dynar_remove_at(peer->current_pieces, piece_index, NULL);
+         remove_current_piece(peer, remote_peer, message->index);
          //Setting the fact that we have the piece
          peer->bitfield[message->index] = '1';
          peer->pieces++;
@@ -456,13 +485,18 @@ void handle_message(peer_t peer, msg_task_t task)
          send_have(peer, message->index);
          //sending UNINTERSTED to peers that doesn't have what we want.
          update_interested_after_receive(peer);
+       } else {                // piece not completed
+         send_request_to_peer(peer, remote_peer, message->index);      // ask for the next block
        }
       } else {
        XBT_DEBUG("However, we already have it");
+       request_new_piece_to_peer(peer, remote_peer);
       }
     }
     break;
   case MESSAGE_CANCEL:
+    XBT_DEBUG("The received CANCEL from %s (%s)",
+             message->mailbox, message->issuer_host_name);
     break;
   }
   //Update the peer speed.
@@ -476,33 +510,28 @@ void handle_message(peer_t peer, msg_task_t task)
   task_message_free(task);
 }
 
-/**
- * Wait for the node to receive interesting bitfield messages (ie: non empty)
- * to be received
- * @param deadline peer deadline
- * @param peer peer data
- */
-void wait_for_pieces(peer_t peer, double deadline)
+void request_new_piece_to_peer(peer_t peer, connection_t remote_peer)
 {
-  int finished = 0;
-  while (MSG_get_clock() < deadline && !finished) {
-    if (peer->comm_received == NULL) {
-      peer->task_received = NULL;
-      peer->comm_received =
-       MSG_task_irecv(&peer->task_received, peer->mailbox);
-    }
-    msg_error_t status = MSG_comm_wait(peer->comm_received, TIMEOUT_MESSAGE);
-    //free the comm already, we don't need it anymore
-    MSG_comm_destroy(peer->comm_received);
-    peer->comm_received = NULL;
-    if (status == MSG_OK) {
-      MSG_task_get_data(peer->task_received);
-      handle_message(peer, peer->task_received);
-      if (peer->current_piece != -1) {
-       finished = 1;
-      }
+  int piece = select_piece_to_download(peer, remote_peer);
+  if (piece != -1) {
+    xbt_dynar_push_as(peer->current_pieces, int, piece);
+    send_request_to_peer(peer, remote_peer, piece);
+  }
+}
+
+void remove_current_piece(peer_t peer, connection_t remote_peer,
+                         int current_piece)
+{
+  int piece_index = -1, piece, i;
+  xbt_dynar_foreach(peer->current_pieces, i, piece) {
+    if (piece == current_piece) {
+      piece_index = i;
+      break;
     }
   }
+  if (piece_index != -1)
+    xbt_dynar_remove_at(peer->current_pieces, piece_index, NULL);
+  remote_peer->current_piece = -1;
 }
 
 /**
@@ -520,53 +549,51 @@ void update_pieces_count_from_bitfield(peer_t peer, char *bitfield)
   }
 }
 
+
+
 /**
- * Update the piece the peer is currently interested in.
- * There is two cases (as described in "Bittorrent Architecture Protocol", Ryan Toole :
- * If the peer has less than 3 pieces, he chooses a piece at random.
+ * Return the piece to be downloaded
+ * There are two cases (as described in "Bittorrent Architecture Protocol", Ryan Toole :
+ * If the peer has strictly less than 4 pieces, he chooses a piece at random.
  * If the peer has more than pieces, he downloads the pieces that are the less
  * replicated
+ * Note: "end game mode" policy is not implemented
+ * @param peer: local peer
+ * @param remote_peer: information about the connection
+ * @return the piece to download if possible. -1 otherwise
  */
-void update_current_piece(peer_t peer)
+int select_piece_to_download(peer_t peer, connection_t remote_peer)
 {
-  if (xbt_dynar_length(peer->current_pieces) >= (FILE_PIECES - peer->pieces)) {
-    return;
+  // TODO : add strict priority policy
+
+  if (xbt_dynar_length(peer->current_pieces) >= (FILE_PIECES - peer->pieces)) {        // end game mode
+    return -1;
   }
-  if (1 || peer->pieces < 3) {
-    int i = 0;
+  if (peer->pieces < 4 && is_interested_and_free(peer, remote_peer)) { // Random first policy
+    int piece = 0;
     do {
-      peer->current_piece =
-       RngStream_RandInt(peer->stream, 0, FILE_PIECES - 1);;
-      i++;
+      piece = RngStream_RandInt(peer->stream, 0, FILE_PIECES - 1);;
     } while (!
-            (peer->bitfield[peer->current_piece] == '0'
-             && !in_current_pieces(peer, peer->current_piece)));
-  } else {
-    //Trivial min algorithm.
-    int i, min_id = -1;
-    short min = -1;
+            (peer->bitfield[piece] == '0'
+             && remote_peer->bitfield[piece] == '1'
+             && !in_current_pieces(peer, piece)));
+    return piece;
+  } else {                     // Rarest first policy
+    int i, min_piece = -1;
+    short min = SHRT_MAX;
     for (i = 0; i < FILE_PIECES; i++) {
-      if (peer->bitfield[i] == '0') {
-       min = peer->pieces_count[i];
-       min_id = i;
-       break;
-      }
-    }
-    xbt_assert((min > -1), "Couldn't find a minimum");
-    for (i = 1; i < FILE_PIECES; i++) {
-      if (peer->pieces_count[i] < min && peer->bitfield[i] == '0') {
+      if (peer->pieces_count[i] < min && peer->bitfield[i] == '0'
+         && remote_peer->bitfield[i] == '1' && !in_current_pieces(peer, i)) {
        min = peer->pieces_count[i];
-       min_id = i;
+       min_piece = i;
       }
     }
-    peer->current_piece = min_id;
+    // TODO: add randomness when selecting a rarest piece
+    return min_piece;
   }
-  xbt_dynar_push_as(peer->current_pieces, int, peer->current_piece);
-  XBT_DEBUG("New interested piece: %d", peer->current_piece);
-  xbt_assert((peer->current_piece >= 0 && peer->current_piece < FILE_PIECES),
-            "Peer want to retrieve a piece that doesn't exist.");
 }
 
+
 /**
  * Update the list of current choked and unchoked peers, using the
  * choke algorithm
@@ -574,34 +601,37 @@ void update_current_piece(peer_t peer)
  */
 void update_choked_peers(peer_t peer)
 {
+  if (nb_interested_peers(peer) == 0)
+    return;
+  //  if(xbt_dict_size(peer->active_peers) > 0)
+  //    return;
+  XBT_DEBUG("(%d) update_choked peers %d active peers", peer->id,
+           xbt_dict_size(peer->active_peers));
   //update the current round
   peer->round = (peer->round + 1) % 3;
-  char *key;
+  char *key, *key_choked;
   connection_t peer_choosed = NULL;
+  connection_t peer_choked = NULL;
   //remove a peer from the list
   xbt_dict_cursor_t cursor = NULL;
   xbt_dict_cursor_first(peer->active_peers, &cursor);
-  if (!xbt_dict_is_empty(peer->active_peers)) {
-    key = xbt_dict_cursor_get_key(cursor);
-    connection_t peer_choked = xbt_dict_cursor_get_data(cursor);
-    if (peer_choked) {
-      send_choked(peer, peer_choked->mailbox);
-      peer_choked->choked_upload = 1;
-    }
-    xbt_dict_remove_ext(peer->active_peers, key, sizeof(int));
+  if (xbt_dict_length(peer->active_peers) > 0) {
+    key_choked = xbt_dict_cursor_get_key(cursor);
+    peer_choked = xbt_dict_cursor_get_data(cursor);
   }
   xbt_dict_cursor_free(&cursor);
 
-       /**
-        * If we are currently seeding, we unchoke the peer which has
-        * been unchoke the least time.
-        */
+  /**
+   * If we are currently seeding, we unchoke the peer which has
+   * been unchoke the least time.
+   */
   if (peer->pieces == FILE_PIECES) {
     connection_t connection;
     double unchoke_time = MSG_get_clock() + 1;
 
     xbt_dict_foreach(peer->peers, cursor, key, connection) {
-      if (connection->last_unchoke < unchoke_time && connection->interested) {
+      if (connection->last_unchoke < unchoke_time && connection->interested
+         && connection->choked_upload) {
        unchoke_time = connection->last_unchoke;
        peer_choosed = connection;
       }
@@ -624,7 +654,7 @@ void update_choked_peers(peer_t peer)
          i++;
        }
        xbt_dict_cursor_free(&cursor);
-       if (peer_choosed->interested == 0) {
+       if (!peer_choosed->interested || !peer_choosed->choked_upload) {
          peer_choosed = NULL;
        }
        j++;
@@ -643,14 +673,45 @@ void update_choked_peers(peer_t peer)
     }
 
   }
-  if (peer_choosed != NULL) {
-    peer_choosed->choked_upload = 0;
-    xbt_dict_set_ext(peer->active_peers, (char *) &peer_choosed->id,
-                    sizeof(int), peer_choosed, NULL);
-    peer_choosed->last_unchoke = MSG_get_clock();
-    send_unchoked(peer, peer_choosed->mailbox);
-  }
+  if (peer_choosed != NULL)
+    XBT_DEBUG
+      ("(%d) update_choked peers unchoked (%d) ; int (%d) ; choked (%d) ",
+       peer->id, peer_choosed->id, peer_choosed->interested,
+       peer_choosed->choked_upload);
 
+  //  if (xbt_dict_size(peer->peers) > 0)
+  //    xbt_assert((xbt_dict_size(peer->active_peers)  != 0),
+  //        "No more active peers !");
+
+
+  //  if (peer_choked != NULL && peer_choked->choked_upload  != 0)
+  //    peer_choked = NULL;
+  //  if (peer_choosed != NULL && peer_choosed->choked_upload  == 0)
+  //    peer_choosed = NULL;
+
+  if (peer_choked != peer_choosed) {
+    if (peer_choked != NULL) {
+      xbt_assert((!peer_choked->choked_upload),
+                "Tries to choked a choked peer");
+      peer_choked->choked_upload = 1;
+      xbt_assert((*((int *) key_choked) == peer_choked->id), "WTF !!!");
+      update_active_peers_set(peer, peer_choked);
+      //      xbt_dict_remove_ext(peer->active_peers, key_choked, sizeof(int));
+      XBT_DEBUG("(%d) Sending a CHOKE to %d", peer->id, peer_choked->id);
+      send_choked(peer, peer_choked->mailbox);
+    }
+    if (peer_choosed != NULL) {
+      xbt_assert((peer_choosed->choked_upload),
+                "Tries to unchoked an unchoked peer");
+      peer_choosed->choked_upload = 0;
+      xbt_dict_set_ext(peer->active_peers, (char *) &peer_choosed->id,
+                      sizeof(int), peer_choosed, NULL);
+      peer_choosed->last_unchoke = MSG_get_clock();
+      XBT_DEBUG("(%d) Sending a UNCHOKE to %d", peer->id, peer_choosed->id);
+      update_active_peers_set(peer, peer_choosed);
+      send_unchoked(peer, peer_choosed->mailbox);
+    }
+  }
 }
 
 /**
@@ -668,15 +729,16 @@ void update_interested_after_receive(peer_t peer)
   xbt_dict_foreach(peer->peers, cursor, key, connection) {
     interested = 0;
     if (connection->am_interested) {
+      xbt_assert(connection->bitfield, "Bitfield not received");
       //Check if the peer still has a piece we want.
-      xbt_dynar_foreach(peer->current_pieces, cpt, piece) {
-       xbt_assert((piece >= 0), "Wrong piece.");
-       if (connection->bitfield && connection->bitfield[piece] == '1') {
+      int i;
+      for (i = 0; i < FILE_PIECES; i++) {
+       if (connection->bitfield[i] == '1' && peer->bitfield[i] == '0') {
          interested = 1;
          break;
        }
       }
-      if (!interested) {
+      if (!interested) {       //no more piece to download from connection
        connection->am_interested = 0;
        send_notinterested(peer, connection->mailbox);
       }
@@ -712,7 +774,8 @@ int piece_complete(peer_t peer, int index)
 }
 
 /**
- * Returns the first block that a peer doesn't have in a piece
+ * Returns the first block that a peer doesn't have in a piece.
+ * If the peer has all blocks of the piece, returns -1.
  */
 int get_first_block(peer_t peer, int piece)
 {
@@ -725,55 +788,103 @@ int get_first_block(peer_t peer, int piece)
   return -1;
 }
 
+/**
+ * Returns a piece that is stored by the remote peer but not by the local peer.
+ * -1 otherwise.
+ */
+int is_interested(peer_t peer, connection_t remote_peer)
+{
+  xbt_assert(remote_peer->bitfield, "Bitfield not received");
+  int i;
+  for (i = 0; i < FILE_PIECES; i++) {
+    if (remote_peer->bitfield[i] == '1' && peer->bitfield[i] == '0') {
+      return 1;
+    }
+  }
+  return 0;
+}
+
+int is_interested_and_free(peer_t peer, connection_t remote_peer)
+{
+  xbt_assert(remote_peer->bitfield, "Bitfield not received");
+  int i;
+  for (i = 0; i < FILE_PIECES; i++) {
+    if (remote_peer->bitfield[i] == '1' && peer->bitfield[i] == '0'
+       && !in_current_pieces(peer, i)) {
+      return 1;
+    }
+  }
+  return 0;
+}
+
+
+/**
+ * Returns a piece that is partially downloaded and stored by the remote peer if any
+ * -1 otherwise.
+ */
+int partially_downloaded_piece(peer_t peer, connection_t remote_peer)
+{
+  xbt_assert(remote_peer->bitfield, "Bitfield not received");
+  int i;
+  for (i = 0; i < FILE_PIECES; i++) {
+    if (remote_peer->bitfield[i] == '1' && peer->bitfield[i] == '0'
+       && !in_current_pieces(peer, i)) {
+      if (get_first_block(peer, i) > 0)
+       return i;
+    }
+  }
+  return -1;
+}
+
+
 /**
  * Send request messages to a peer that have unchoked us
  * @param peer peer
  * @param remote_peer peer data to the peer we want to send the request
  */
-void send_requests_to_peer(peer_t peer, connection_t remote_peer)
+void send_request_to_peer(peer_t peer, connection_t remote_peer, int piece)
 {
+  remote_peer->current_piece = piece;
   unsigned i;
-  int piece, block_index, block_length;
-  xbt_dynar_foreach(peer->current_pieces, i, piece) {
-    if (remote_peer->bitfield && remote_peer->bitfield[piece] == '1') {
-      block_index = get_first_block(peer, piece);
-      if (block_index != -1) {
-       block_length = PIECES_BLOCKS - block_index;
-       block_length = min(BLOCKS_REQUESTED, block_length);
-       send_request(peer, remote_peer->mailbox, piece, block_index,
-                    block_length);
-       break;
-      }
-    }
+  int block_index, block_length;
+  xbt_assert(remote_peer->bitfield, "bitfield not received");
+  xbt_assert(remote_peer->bitfield[piece] == '1', "WTF !!!");
+  block_index = get_first_block(peer, piece);
+  if (block_index != -1) {
+    block_length = PIECES_BLOCKS - block_index;
+    block_length = min(BLOCKS_REQUESTED, block_length);
+    send_request(peer, remote_peer->mailbox, piece, block_index,
+                block_length);
   }
 }
 
+
 /**
- * Find the peers that have the current interested piece and send them
- * the "interested" message
+ * Indicates if a piece is currently being downloaded by the peer.
  */
-void send_interested_to_peers(peer_t peer)
+int in_current_pieces(peer_t peer, int piece)
 {
-  char *key;
-  xbt_dict_cursor_t cursor = NULL;
-  connection_t connection;
-  xbt_assert((peer->current_piece != -1),
-            "Tried to send a interested message wheras the current_piece is -1");
-  xbt_dict_foreach(peer->peers, cursor, key, connection) {
-    if (connection->bitfield
-       && connection->bitfield[peer->current_piece] == '1') {
-      connection->am_interested = 1;
-      msg_task_t task =
-       task_message_new(MESSAGE_INTERESTED, peer->hostname, peer->mailbox,
-                        peer->id, task_message_size(MESSAGE_INTERESTED));
-      MSG_task_dsend(task, connection->mailbox, task_message_free);
-      XBT_DEBUG("Send INTERESTED to %s", connection->mailbox);
+  unsigned i;
+  int peer_piece;
+  xbt_dynar_foreach(peer->current_pieces, i, peer_piece) {
+    if (peer_piece == piece) {
+      return 1;
     }
   }
-  peer->current_piece = -1;
-  peer->pieces_requested++;
+  return 0;
 }
 
+
+
+
+/***********************************************************
+ *
+ *  Low level message functions
+ *
+ ***********************************************************/
+
+
+
 /**
  * Send a "interested" message to a peer
  * @param peer peer data
@@ -923,16 +1034,3 @@ void send_piece(peer_t peer, const char *mailbox, int piece, int stalled,
                           stalled, block_index, block_length, BLOCK_SIZE);
   MSG_task_dsend(task, mailbox, task_message_free);
 }
-
-int in_current_pieces(peer_t peer, int piece)
-{
-  unsigned i;
-  int is_in = 0, peer_piece;
-  xbt_dynar_foreach(peer->current_pieces, i, peer_piece) {
-    if (peer_piece == piece) {
-      is_in = 1;
-      break;
-    }
-  }
-  return is_in;
-}