Logo AND Algorithmique Numérique Distribuée

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