Logo AND Algorithmique Numérique Distribuée

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