Logo AND Algorithmique Numérique Distribuée

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