Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
Update copyright lines for 2023.
[simgrid.git] / examples / c / dht-kademlia / node.c
1 /* Copyright (c) 2010-2023. 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 "node.h"
7 #include "routing_table.h"
8 #include "simgrid/comm.h"
9
10 #include <stdio.h> /* snprintf */
11
12 XBT_LOG_NEW_DEFAULT_CATEGORY(dht_kademlia_node, "Messages specific for this example");
13
14 /** @brief Initialization of a node
15  * @param node_id the id of the node
16  * @return the node created
17  */
18 node_t node_init(unsigned int node_id)
19 {
20   node_t node             = xbt_new(s_node_t, 1);
21   node->id                = node_id;
22   node->table             = routing_table_init(node_id);
23   node->mailbox           = get_node_mailbox(node_id);
24   node->find_node_failed  = 0;
25   node->find_node_success = 0;
26
27   node->received_msg = NULL;
28   node->receive_comm = NULL;
29
30   return node;
31 }
32
33 /* @brief Node destructor  */
34 void node_free(node_t node)
35 {
36   routing_table_free(node->table);
37   xbt_free(node);
38 }
39
40 /**
41   * Try to asynchronously get a new message from given mailbox. Return null if none available.
42   */
43 kademlia_message_t receive(node_t node, sg_mailbox_t mailbox)
44 {
45   if (node->receive_comm == NULL)
46     node->receive_comm = sg_mailbox_get_async(mailbox, &node->received_msg);
47   if (!sg_comm_test(node->receive_comm))
48     return NULL;
49   node->receive_comm = NULL;
50   return node->received_msg;
51 }
52
53 /**
54  * @brief Tries to join the network
55  * @param node node data
56  * @param id_known id of the node I know in the network.
57  */
58 unsigned int join(node_t node, unsigned int id_known)
59 {
60   unsigned int i;
61   unsigned int got_answer = 0;
62
63   sg_mailbox_t mailbox = get_node_mailbox(node->id);
64
65   /* Add the guy we know to our routing table and ourselves. */
66   routing_table_update(node, node->id);
67   routing_table_update(node, id_known);
68
69   /* First step: Send a "FIND_NODE" request to the node we know */
70   send_find_node(node, id_known, node->id);
71   do {
72     const kademlia_message_t msg = receive(node, mailbox);
73     if (msg) {
74       XBT_DEBUG("Received an answer from the node I know.");
75       got_answer = 1;
76       // retrieve the node list and ping them.
77       const s_answer_t* node_list  = msg->answer;
78       if (node_list != NULL) {
79         node_contact_t contact;
80         xbt_dynar_foreach (node_list->nodes, i, contact)
81           routing_table_update(node, contact->id);
82         node->received_msg = NULL;
83       } else {
84         handle_find_node(node, msg);
85       }
86       free_message(msg);
87     } else {
88       sg_actor_sleep_for(1);
89     }
90   } while (got_answer == 0);
91
92   /* Second step: Send a FIND_NODE to a random node in buckets */
93   unsigned int bucket_id = routing_table_find_bucket(node->table, id_known)->id;
94   xbt_assert(bucket_id <= IDENTIFIER_SIZE);
95   for (i = 0; ((bucket_id > i) || (bucket_id + i) <= IDENTIFIER_SIZE) && i < JOIN_BUCKETS_QUERIES; i++) {
96     if (bucket_id > i) {
97       unsigned int id_in_bucket = get_id_in_prefix(node->id, bucket_id - i);
98       find_node(node, id_in_bucket, 0);
99     }
100     if (bucket_id + i <= IDENTIFIER_SIZE) {
101       unsigned int id_in_bucket = get_id_in_prefix(node->id, bucket_id + i);
102       find_node(node, id_in_bucket, 0);
103     }
104   }
105   return got_answer;
106 }
107
108 /** @brief Send a "FIND_NODE" to a node
109  * @param node sender node data
110  * @param id node we are querying
111  * @param destination node we are trying to find.
112  */
113 void send_find_node(const_node_t node, unsigned int id, unsigned int destination)
114 {
115   /* Gets the mailbox to send to */
116   sg_mailbox_t mailbox = get_node_mailbox(id);
117   /* Build the message */
118   kademlia_message_t msg = new_message(node->id, destination, NULL, node->mailbox, sg_host_self_get_name());
119   sg_comm_t comm         = sg_mailbox_put_init(mailbox, msg, COMM_SIZE);
120   sg_comm_detach(comm, free_message);
121   XBT_VERB("Asking %u for its closest nodes", id);
122 }
123
124 /**
125  * Sends to the best "KADEMLIA_ALPHA" nodes in the "node_list" array a "FIND_NODE" request, to ask them for their best
126  * nodes
127  */
128 unsigned int send_find_node_to_best(const_node_t node, const_answer_t node_list)
129 {
130   unsigned int i           = 0;
131   unsigned int j           = 0;
132   unsigned int destination = node_list->destination_id;
133   while (j < KADEMLIA_ALPHA && i < node_list->size) {
134     /* We need to have at most "KADEMLIA_ALPHA" requests each time, according to the protocol */
135     /* Gets the node we want to send the query to */
136     const s_node_contact_t* node_to_query = xbt_dynar_get_as(node_list->nodes, i, node_contact_t);
137     if (node_to_query->id != node->id) { /* No need to query ourselves */
138       send_find_node(node, node_to_query->id, destination);
139       j++;
140     }
141     i++;
142   }
143   return i;
144 }
145
146 /** @brief Updates/Puts the node id unsigned into our routing table
147  * @param node Our node data
148  * @param id The id of the node we need to add unsigned into our routing table
149  */
150 void routing_table_update(const_node_t node, unsigned int id)
151 {
152   const_routing_table_t table = node->table;
153   // retrieval of the bucket in which the should be
154   const_bucket_t bucket = routing_table_find_bucket(table, id);
155
156   // check if the id is already in the bucket.
157   unsigned int id_pos = bucket_find_id(bucket, id);
158
159   if (id_pos == (unsigned int)-1) {
160     /* We check if the bucket is full or not. If it is, we evict an old element */
161     if (xbt_dynar_length(bucket->nodes) >= BUCKET_SIZE)
162       xbt_dynar_pop(bucket->nodes, NULL);
163
164     xbt_dynar_unshift(bucket->nodes, &id);
165     XBT_VERB("I'm adding to my routing table %08x", id);
166   } else {
167     // We push to the front of the dynar the element.
168     unsigned int element = xbt_dynar_get_as(bucket->nodes, id_pos, unsigned int);
169     xbt_dynar_remove_at(bucket->nodes, id_pos, NULL);
170     xbt_dynar_unshift(bucket->nodes, &element);
171     XBT_VERB("I'm updating %08x", element);
172   }
173 }
174
175 /** @brief Finds the closest nodes to the node given.
176  * @param node : our node
177  * @param destination_id : the id of the guy we are trying to find
178  */
179 answer_t find_closest(const_node_t node, unsigned int destination_id)
180 {
181   answer_t answer = answer_init(destination_id);
182   /* We find the corresponding bucket for the id */
183   const_bucket_t bucket = routing_table_find_bucket(node->table, destination_id);
184   int bucket_id         = bucket->id;
185   xbt_assert((bucket_id <= IDENTIFIER_SIZE), "Bucket found has a wrong identifier");
186   /* So, we copy the contents of the bucket unsigned into our result dynar */
187   answer_add_bucket(bucket, answer);
188
189   /* However, if we don't have enough elements in our bucket, we NEED to include at least "BUCKET_SIZE" elements
190    * (if, of course, we know at least "BUCKET_SIZE" elements. So we're going to look unsigned into the other buckets.
191    */
192   for (int i = 1; answer->size < BUCKET_SIZE && ((bucket_id - i > 0) || (bucket_id + i < IDENTIFIER_SIZE)); i++) {
193     /* We check the previous buckets */
194     if (bucket_id - i >= 0) {
195       const_bucket_t bucket_p = &node->table->buckets[bucket_id - i];
196       answer_add_bucket(bucket_p, answer);
197     }
198     /* We check the next buckets */
199     if (bucket_id + i <= IDENTIFIER_SIZE) {
200       const_bucket_t bucket_n = &node->table->buckets[bucket_id + i];
201       answer_add_bucket(bucket_n, answer);
202     }
203   }
204   /* We sort the array by XOR distance */
205   answer_sort(answer);
206   /* We trim the array to have only BUCKET_SIZE or less elements */
207   answer_trim(answer);
208
209   return answer;
210 }
211
212 unsigned int find_node(node_t node, unsigned int id_to_find, unsigned int count_in_stats)
213 {
214   unsigned int queries;
215   unsigned int answers;
216   unsigned int destination_found = 0;
217   unsigned int nodes_added       = 0;
218   double global_timeout          = simgrid_get_clock() + FIND_NODE_GLOBAL_TIMEOUT;
219   unsigned int steps             = 0;
220
221   /* First we build a list of who we already know */
222   answer_t node_list = find_closest(node, id_to_find);
223   xbt_assert((node_list != NULL), "node_list incorrect");
224
225   XBT_DEBUG("Doing a FIND_NODE on %08x", id_to_find);
226
227   /* Ask the nodes on our list if they   have information about the node we are trying to find */
228   sg_mailbox_t mailbox = get_node_mailbox(node->id);
229   do {
230     answers        = 0;
231     queries        = send_find_node_to_best(node, node_list);
232     nodes_added    = 0;
233     double timeout = simgrid_get_clock() + FIND_NODE_TIMEOUT;
234     steps++;
235     double time_beginreceive = simgrid_get_clock();
236
237     do {
238       const kademlia_message_t msg = receive(node, mailbox);
239       if (msg) {
240         // Figure out if we received an answer or something else
241         // Check if what we have received is what we are looking for.
242         if (msg->answer != NULL && msg->answer->destination_id == id_to_find) {
243           // Handle the answer
244           routing_table_update(node, msg->sender_id);
245           node_contact_t contact;
246           unsigned int i;
247           xbt_dynar_foreach (node_list->nodes, i, contact)
248             routing_table_update(node, contact->id);
249
250           answers++;
251
252           nodes_added = answer_merge(node_list, msg->answer);
253           XBT_DEBUG("Received an answer from %s (%s) with %lu nodes on it", sg_mailbox_get_name(msg->answer_to),
254                     msg->issuer_host_name, xbt_dynar_length(msg->answer->nodes));
255         } else {
256           if (msg->answer != NULL) {
257             routing_table_update(node, msg->sender_id);
258             XBT_DEBUG("Received a wrong answer for a FIND_NODE");
259           } else {
260             handle_find_node(node, msg);
261           }
262           // Update the timeout if we didn't have our answer
263           timeout += simgrid_get_clock() - time_beginreceive;
264           time_beginreceive = simgrid_get_clock();
265         }
266         free_message(msg);
267       } else {
268         sg_actor_sleep_for(1);
269       }
270     } while (simgrid_get_clock() < timeout && answers < queries);
271     destination_found = answer_destination_found(node_list);
272   } while (!destination_found && (nodes_added > 0 || answers == 0) && simgrid_get_clock() < global_timeout &&
273            steps < MAX_STEPS);
274   if (destination_found) {
275     if (count_in_stats)
276       node->find_node_success++;
277     if (queries > 4)
278       XBT_VERB("FIND_NODE on %08x success in %u steps", id_to_find, steps);
279     routing_table_update(node, id_to_find);
280   } else {
281     if (count_in_stats) {
282       node->find_node_failed++;
283       XBT_VERB("%08x not found in %u steps", id_to_find, steps);
284     }
285   }
286   answer_free(node_list);
287   return destination_found;
288 }
289
290 /** @brief Does a pseudo-random lookup for someone in the system
291  * @param node caller node data
292  */
293 void random_lookup(node_t node)
294 {
295   unsigned int id_to_look = RANDOM_LOOKUP_NODE; // Totally random.
296   /* TODO: Use some pseudorandom generator. */
297   XBT_DEBUG("I'm doing a random lookup");
298   find_node(node, id_to_look, 1);
299 }
300
301 /** @brief Handles the answer to an incoming "find_node" message */
302 void handle_find_node(const_node_t node, const_kademlia_message_t msg)
303 {
304   routing_table_update(node, msg->sender_id);
305   XBT_VERB("Received a FIND_NODE from %s (%s), he's trying to find %08x", sg_mailbox_get_name(msg->answer_to),
306            msg->issuer_host_name, msg->destination_id);
307   // Building the msg to send
308   kademlia_message_t answer = new_message(node->id, msg->destination_id, find_closest(node, msg->destination_id),
309                                           node->mailbox, sg_host_self_get_name());
310   // Sending the msg
311   sg_comm_t comm = sg_mailbox_put_init(msg->answer_to, answer, COMM_SIZE);
312   sg_comm_detach(comm, &free_message);
313 }
314
315 /**@brief Returns an identifier which is in a specific bucket of a routing table
316  * @param id id of the routing table owner
317  * @param prefix id of the bucket where we want that identifier to be
318  */
319 unsigned int get_id_in_prefix(unsigned int id, unsigned int prefix)
320 {
321   if (prefix == 0) {
322     return 0;
323   } else {
324     return (1U << (prefix - 1)) ^ id;
325   }
326 }
327
328 /** @brief Returns the prefix of an identifier.
329  * The prefix is the id of the bucket in which the remote identifier xor our identifier should be stored.
330  * @param id : big unsigned int id to test
331  * @param nb_bits : key size
332  */
333 unsigned int get_node_prefix(unsigned int id, unsigned int nb_bits)
334 {
335   unsigned int size = sizeof(unsigned int) * 8;
336   for (unsigned int j = 0; j < size; j++) {
337     if (((id >> (size - 1 - j)) & 0x1) != 0) {
338       return nb_bits - j;
339     }
340   }
341   return 0;
342 }
343
344 /** @brief Gets the mailbox name of a host given its identifier */
345 sg_mailbox_t get_node_mailbox(unsigned int id)
346 {
347   char mailbox_name[MAILBOX_NAME_SIZE];
348   snprintf(mailbox_name, MAILBOX_NAME_SIZE - 1, "%u", id);
349   return sg_mailbox_by_name(mailbox_name);
350 }
351
352 /** Constructor, build a new contact information. */
353 node_contact_t node_contact_new(unsigned int id, unsigned int distance)
354 {
355   node_contact_t contact = xbt_new(s_node_contact_t, 1);
356
357   contact->id       = id;
358   contact->distance = distance;
359
360   return contact;
361 }
362
363 /** Builds a contact information from a contact information */
364 node_contact_t node_contact_copy(const_node_contact_t node_contact)
365 {
366   node_contact_t contact = xbt_new(s_node_contact_t, 1);
367
368   contact->id       = node_contact->id;
369   contact->distance = node_contact->distance;
370
371   return contact;
372 }
373
374 /** Destructor */
375 void node_contact_free(node_contact_t contact)
376 {
377   xbt_free(contact);
378 }