Logo AND Algorithmique Numérique Distribuée

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