Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
Another step toward working CommPtr. chord example is broken ATM
[simgrid.git] / examples / s4u / dht-chord / node.cpp
1 /* Copyright (c) 2010-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 "s4u_dht-chord.hpp"
7
8 XBT_LOG_EXTERNAL_DEFAULT_CATEGORY(s4u_chord);
9
10 /* Returns whether an id belongs to the interval [start, end].
11  *
12  * The parameters are normalized to make sure they are between 0 and nb_keys - 1).
13  * 1 belongs to [62, 3]
14  * 1 does not belong to [3, 62]
15  * 63 belongs to [62, 3]
16  * 63 does not belong to [3, 62]
17  * 24 belongs to [21, 29]
18  * 24 does not belong to [29, 21]
19  *
20  * \param id id to check
21  * \param start lower bound
22  * \param end upper bound
23  * \return a non-zero value if id in in [start, end]
24  */
25 static int is_in_interval(int id, int start, int end)
26 {
27   int i = id % nb_keys;
28   int s = start % nb_keys;
29   int e = end % nb_keys;
30
31   // make sure end >= start and id >= start
32   if (e < s) {
33     e += nb_keys;
34   }
35
36   if (i < s) {
37     i += nb_keys;
38   }
39
40   return i <= e;
41 }
42
43 /* Initializes the current node as the first one of the system */
44 Node::Node(std::vector<std::string> args)
45 {
46   xbt_assert(args.size() == 3 || args.size() == 5, "Wrong number of arguments for this node");
47
48   // initialize my node
49   id_                = std::stoi(args[1]);
50   stream             = simgrid::s4u::this_actor::host()->extension<HostChord>()->getStream();
51   mailbox_           = simgrid::s4u::Mailbox::byName(std::to_string(id_));
52   next_finger_to_fix = 0;
53   fingers_           = new int[nb_bits];
54
55   for (int i = 0; i < nb_bits; i++) {
56     fingers_[i] = id_;
57   }
58
59   if (args.size() == 3) { // first ring
60     deadline_   = std::stod(args[2]);
61     start_time_ = simgrid::s4u::Engine::getClock();
62     XBT_DEBUG("Create a new Chord ring...");
63   } else {
64     known_id_   = std::stoi(args[2]);
65     start_time_ = std::stod(args[3]);
66     deadline_   = std::stod(args[4]);
67     XBT_DEBUG("Hey! Let's join the system in %f seconds (shall leave at time %f)", start_time_,
68               start_time_ + deadline_);
69   }
70 }
71
72 Node::~Node()
73 {
74   delete[] fingers_;
75 }
76 /* Makes the current node join the ring, knowing the id of a node already in the ring
77  *
78  * \param known_id id of a node already in the ring
79  * \return true if the join operation succeeded
80  *  */
81
82 void Node::join(int known_id)
83 {
84   XBT_INFO("Joining the ring with id %d, knowing node %d", id_, known_id);
85   setPredecessor(-1); // no predecessor (yet)
86
87   int successor_id = remoteFindSuccessor(known_id, id_);
88   if (successor_id == -1) {
89     XBT_INFO("Cannot join the ring.");
90   } else {
91     setFinger(0, successor_id);
92     printFingerTable();
93     joined = true;
94   }
95 }
96
97 /* Makes the current node quit the system */
98 void Node::leave()
99 {
100   XBT_INFO("Well Guys! I Think it's time for me to leave ;)");
101   notifyAndQuit();
102   joined = false;
103 }
104
105 /* Notifies the successor and the predecessor of the current node before leaving */
106 void Node::notifyAndQuit()
107 {
108   // send the PREDECESSOR_LEAVING to our successor
109   ChordMessage* pred_msg = new ChordMessage(PREDECESSOR_LEAVING);
110   pred_msg->request_id   = pred_id_;
111   pred_msg->answer_to    = mailbox_;
112
113   XBT_DEBUG("Sending a 'PREDECESSOR_LEAVING' to my successor %d", fingers_[0]);
114   try {
115     simgrid::s4u::this_actor::send(simgrid::s4u::Mailbox::byName(std::to_string(fingers_[0])), pred_msg, 10, timeout);
116   } catch (xbt_ex& e) {
117     if (e.category == timeout_error) {
118       XBT_DEBUG("Timeout expired when sending a 'PREDECESSOR_LEAVING' to my successor %d", fingers_[0]);
119       delete pred_msg;
120     }
121   }
122
123   // send the SUCCESSOR_LEAVING to our predecessor
124   ChordMessage* succ_msg = new ChordMessage(SUCCESSOR_LEAVING);
125   succ_msg->request_id   = fingers_[0];
126   succ_msg->answer_to    = mailbox_;
127   XBT_DEBUG("Sending a 'SUCCESSOR_LEAVING' to my predecessor %d", pred_id_);
128
129   try {
130     simgrid::s4u::this_actor::send(simgrid::s4u::Mailbox::byName(std::to_string(pred_id_)), succ_msg, 10, timeout);
131   } catch (xbt_ex& e) {
132     if (e.category == timeout_error) {
133       XBT_DEBUG("Timeout expired when sending a 'SUCCESSOR_LEAVING' to my predecessor %d", pred_id_);
134       delete succ_msg;
135     }
136   }
137 }
138
139 /* Performs a find successor request to a random id */
140 void Node::randomLookup()
141 {
142   int res          = id_;
143   int random_index = RngStream_RandInt(stream, 0, nb_bits - 1);
144   int random_id    = fingers_[random_index];
145   XBT_DEBUG("Making a lookup request for id %d", random_id);
146   if (random_id != id_)
147     res = findSuccessor(random_id);
148   XBT_DEBUG("The successor of node %d is %d", random_id, res);
149 }
150
151 /* Sets a finger of the current node.
152  *
153  * \param node the current node
154  * \param finger_index index of the finger to set (0 to nb_bits - 1)
155  * \param id the id to set for this finger
156  */
157 void Node::setFinger(int finger_index, int id)
158 {
159   if (id != fingers_[finger_index]) {
160     fingers_[finger_index] = id;
161     XBT_VERB("My new finger #%d is %d", finger_index, id);
162   }
163 }
164
165 /* Sets the predecessor of the current node.
166  * \param id the id to predecessor, or -1 to unset the predecessor
167  */
168 void Node::setPredecessor(int predecessor_id)
169 {
170   if (predecessor_id != pred_id_) {
171     pred_id_ = predecessor_id;
172     XBT_VERB("My new predecessor is %d", predecessor_id);
173   }
174 }
175
176 /** refreshes the finger table of the current node (called periodically) */
177 void Node::fixFingers()
178 {
179   XBT_DEBUG("Fixing fingers");
180   int id = findSuccessor(id_ + powers2[next_finger_to_fix]);
181   if (id != -1) {
182     if (id != fingers_[next_finger_to_fix]) {
183       setFinger(next_finger_to_fix, id);
184       printFingerTable();
185     }
186     next_finger_to_fix = (next_finger_to_fix + 1) % nb_bits;
187   }
188 }
189
190 /** Displays the finger table of a node. */
191 void Node::printFingerTable()
192 {
193   if (XBT_LOG_ISENABLED(s4u_chord, xbt_log_priority_verbose)) {
194     XBT_VERB("My finger table:");
195     XBT_VERB("Start | Succ");
196     for (int i = 0; i < nb_bits; i++) {
197       XBT_VERB(" %3d  | %3d", (id_ + powers2[i]) % nb_keys, fingers_[i]);
198     }
199
200     XBT_VERB("Predecessor: %d", pred_id_);
201   }
202 }
203
204 /* checks whether the predecessor has failed (called periodically) */
205 void Node::checkPredecessor()
206 {
207   XBT_DEBUG("Checking whether my predecessor is alive");
208   void* data = nullptr;
209   if (pred_id_ == -1)
210     return;
211
212   simgrid::s4u::MailboxPtr mailbox        = simgrid::s4u::Mailbox::byName(std::to_string(pred_id_));
213   simgrid::s4u::MailboxPtr return_mailbox = simgrid::s4u::Mailbox::byName(std::to_string(id_) + "_is_alive");
214
215   ChordMessage* message = new ChordMessage(PREDECESSOR_ALIVE);
216   message->request_id   = pred_id_;
217   message->answer_to    = return_mailbox;
218
219   XBT_DEBUG("Sending a 'Predecessor Alive' request to my predecessor %d", pred_id_);
220   try {
221     simgrid::s4u::this_actor::send(mailbox, message, 10, timeout);
222   } catch (xbt_ex& e) {
223     if (e.category == timeout_error) {
224       XBT_DEBUG("Failed to send the 'Predecessor Alive' request to %d", pred_id_);
225       delete message;
226       return;
227     }
228   }
229   // receive the answer
230   XBT_DEBUG("Sent 'Predecessor Alive' request to %d, waiting for the answer on my mailbox '%s'", pred_id_,
231             message->answer_to->name());
232   simgrid::s4u::CommPtr comm = simgrid::s4u::this_actor::irecv(return_mailbox, &data);
233
234   try {
235     comm->wait(timeout);
236     XBT_DEBUG("Received the answer to my 'Predecessor Alive': my predecessor %d is alive", pred_id_);
237     delete message;
238   } catch (xbt_ex& e) {
239     if (e.category == timeout_error) {
240       XBT_DEBUG("Failed to receive the answer to my 'Predecessor Alive' request");
241       pred_id_ = -1;
242     }
243   }
244 }
245
246 /* Asks its predecessor to a remote node
247  *
248  * \param ask_to the node to ask to
249  * \return the id of its predecessor node, or -1 if the request failed (or if the node does not know its predecessor)
250  */
251 int Node::remoteGetPredecessor(int ask_to)
252 {
253   int predecessor_id                      = -1;
254   void* data                              = nullptr;
255   simgrid::s4u::MailboxPtr mailbox        = simgrid::s4u::Mailbox::byName(std::to_string(ask_to));
256   simgrid::s4u::MailboxPtr return_mailbox = simgrid::s4u::Mailbox::byName(std::to_string(id_) + "_pred");
257
258   ChordMessage* message = new ChordMessage(GET_PREDECESSOR);
259   message->request_id   = id_;
260   message->answer_to    = return_mailbox;
261
262   // send a "Get Predecessor" request to ask_to_id
263   XBT_DEBUG("Sending a 'Get Predecessor' request to %d", ask_to);
264   try {
265     simgrid::s4u::this_actor::send(mailbox, message, 10, timeout);
266   } catch (xbt_ex& e) {
267     if (e.category == timeout_error) {
268       XBT_DEBUG("Failed to send the 'Get Predecessor' request to %d", ask_to);
269       delete message;
270       return predecessor_id;
271     }
272   }
273
274   // receive the answer
275   XBT_DEBUG("Sent 'Get Predecessor' request to %d, waiting for the answer on my mailbox '%s'", ask_to,
276             message->answer_to->name());
277   simgrid::s4u::CommPtr comm = simgrid::s4u::this_actor::irecv(return_mailbox, &data);
278
279   try {
280     comm->wait(timeout);
281     ChordMessage* answer = static_cast<ChordMessage*>(data);
282     XBT_DEBUG("Received the answer to my 'Get Predecessor' request: the predecessor of node %d is %d", ask_to,
283               answer->answer_id);
284     predecessor_id = answer->answer_id;
285     delete answer;
286   } catch (xbt_ex& e) {
287     if (e.category == timeout_error) {
288       XBT_DEBUG("Failed to receive the answer to my 'Get Predecessor' request");
289       delete static_cast<ChordMessage*>(data);
290     }
291   }
292
293   return predecessor_id;
294 }
295
296 /* Returns the closest preceding finger of an id with respect to the finger table of the current node.
297  *
298  * \param id the id to find
299  * \return the closest preceding finger of that id
300  */
301 int Node::closestPrecedingFinger(int id)
302 {
303   for (int i = nb_bits - 1; i >= 0; i--) {
304     if (is_in_interval(fingers_[i], id_ + 1, id - 1)) {
305       return fingers_[i];
306     }
307   }
308   return id_;
309 }
310
311 /* Makes the current node find the successor node of an id.
312  *
313  * \param id the id to find
314  * \return the id of the successor node, or -1 if the request failed
315  */
316 int Node::findSuccessor(int id)
317 {
318   // is my successor the successor?
319   if (is_in_interval(id, id_ + 1, fingers_[0])) {
320     return fingers_[0];
321   }
322
323   // otherwise, ask the closest preceding finger in my table
324   return remoteFindSuccessor(closestPrecedingFinger(id), id);
325 }
326
327 int Node::remoteFindSuccessor(int ask_to, int id)
328 {
329   int successor                           = -1;
330   void* data                              = nullptr;
331   simgrid::s4u::MailboxPtr mailbox        = simgrid::s4u::Mailbox::byName(std::to_string(ask_to));
332   simgrid::s4u::MailboxPtr return_mailbox = simgrid::s4u::Mailbox::byName(std::to_string(id_) + "_succ");
333
334   ChordMessage* message = new ChordMessage(FIND_SUCCESSOR);
335   message->request_id   = id_;
336   message->answer_to    = return_mailbox;
337
338   // send a "Find Successor" request to ask_to_id
339   XBT_DEBUG("Sending a 'Find Successor' request to %d for id %d", ask_to, id);
340   try {
341     simgrid::s4u::this_actor::send(mailbox, message, 10, timeout);
342   } catch (xbt_ex& e) {
343     if (e.category == timeout_error) {
344       XBT_DEBUG("Failed to send the 'Find Successor' request to %d for id %d", ask_to, id_);
345       delete message;
346       return successor;
347     }
348   }
349   // receive the answer
350   XBT_DEBUG("Sent a 'Find Successor' request to %d for key %d, waiting for the answer", ask_to, id);
351   simgrid::s4u::CommPtr comm = simgrid::s4u::this_actor::irecv(return_mailbox, &data);
352
353   try {
354     comm->wait(timeout);
355     ChordMessage* answer = static_cast<ChordMessage*>(data);
356     XBT_DEBUG("Received the answer to my 'Find Successor' request for id %d: the successor of key %d is %d",
357               answer->request_id, id_, answer->answer_id);
358     successor = answer->answer_id;
359     delete answer;
360   } catch (xbt_ex& e) {
361     if (e.category == timeout_error) {
362       XBT_DEBUG("Failed to receive the answer to my 'Find Successor' request");
363       delete static_cast<ChordMessage*>(data);
364     }
365   }
366   return successor;
367 }
368
369 /* Notifies the current node that its predecessor may have changed. */
370 void Node::notify(int predecessor_candidate_id)
371 {
372   if (pred_id_ == -1 || is_in_interval(predecessor_candidate_id, pred_id_ + 1, id_ - 1)) {
373     setPredecessor(predecessor_candidate_id);
374     printFingerTable();
375   } else {
376     XBT_DEBUG("I don't have to change my predecessor to %d", predecessor_candidate_id);
377   }
378 }
379
380 /* Notifies a remote node that its predecessor may have changed. */
381 void Node::remoteNotify(int notify_id, int predecessor_candidate_id)
382 {
383   ChordMessage* message = new ChordMessage(NOTIFY);
384   message->request_id   = predecessor_candidate_id;
385   message->answer_to    = nullptr;
386
387   // send a "Notify" request to notify_id
388   XBT_DEBUG("Sending a 'Notify' request to %d", notify_id);
389   simgrid::s4u::MailboxPtr mailbox = simgrid::s4u::Mailbox::byName(std::to_string(notify_id));
390   simgrid::s4u::this_actor::dsend(mailbox, message, 10);
391 }
392
393 /* This function is called periodically. It checks the immediate successor of the current node. */
394 void Node::stabilize()
395 {
396   XBT_DEBUG("Stabilizing node");
397
398   // get the predecessor of my immediate successor
399   int candidate_id;
400   int successor_id = fingers_[0];
401   if (successor_id != id_) {
402     candidate_id = remoteGetPredecessor(successor_id);
403   } else {
404     candidate_id = pred_id_;
405   }
406
407   // this node is a candidate to become my new successor
408   if (candidate_id != -1 && is_in_interval(candidate_id, id_ + 1, successor_id - 1)) {
409     setFinger(0, candidate_id);
410   }
411   if (successor_id != id_) {
412     remoteNotify(successor_id, id_);
413   }
414 }
415
416 /* This function is called when a node receives a message.
417  *
418  * \param message the message to handle (don't touch it afterward: it will be destroyed, reused or forwarded)
419  */
420 void Node::handleMessage(ChordMessage* message)
421 {
422   switch (message->type) {
423   case FIND_SUCCESSOR:
424     XBT_DEBUG("Received a 'Find Successor' request from %s for id %d", message->issuer_host_name.c_str(),
425         message->request_id);
426     // is my successor the successor?
427     if (is_in_interval(message->request_id, id_ + 1, fingers_[0])) {
428       message->type = FIND_SUCCESSOR_ANSWER;
429       message->answer_id = fingers_[0];
430       XBT_DEBUG("Sending back a 'Find Successor Answer' to %s (mailbox %s): the successor of %d is %d",
431           message->issuer_host_name.c_str(), message->answer_to->name(), message->request_id, message->answer_id);
432       simgrid::s4u::this_actor::dsend(message->answer_to, message, 10);
433     } else {
434       // otherwise, forward the request to the closest preceding finger in my table
435       int closest = closestPrecedingFinger(message->request_id);
436       XBT_DEBUG("Forwarding the 'Find Successor' request for id %d to my closest preceding finger %d",
437           message->request_id, closest);
438       simgrid::s4u::MailboxPtr mailbox = simgrid::s4u::Mailbox::byName(std::to_string(closest));
439       simgrid::s4u::this_actor::dsend(mailbox, message, 10);
440     }
441     break;
442
443   case GET_PREDECESSOR:
444     XBT_DEBUG("Receiving a 'Get Predecessor' request from %s", message->issuer_host_name.c_str());
445     message->type = GET_PREDECESSOR_ANSWER;
446     message->answer_id = pred_id_;
447     XBT_DEBUG("Sending back a 'Get Predecessor Answer' to %s via mailbox '%s': my predecessor is %d",
448         message->issuer_host_name.c_str(), message->answer_to->name(), message->answer_id);
449     simgrid::s4u::this_actor::dsend(message->answer_to, message, 10);
450     break;
451
452   case NOTIFY:
453     // someone is telling me that he may be my new predecessor
454     XBT_DEBUG("Receiving a 'Notify' request from %s", message->issuer_host_name.c_str());
455     notify(message->request_id);
456     delete message;
457     break;
458
459   case PREDECESSOR_LEAVING:
460     // my predecessor is about to quit
461     XBT_DEBUG("Receiving a 'Predecessor Leaving' message from %s", message->issuer_host_name.c_str());
462     // modify my predecessor
463     setPredecessor(message->request_id);
464     delete message;
465     /*TODO :
466       >> notify my new predecessor
467       >> send a notify_predecessors !!
468      */
469     break;
470
471   case SUCCESSOR_LEAVING:
472     // my successor is about to quit
473     XBT_DEBUG("Receiving a 'Successor Leaving' message from %s", message->issuer_host_name.c_str());
474     // modify my successor FIXME : this should be implicit ?
475     setFinger(0, message->request_id);
476     delete message;
477     /* TODO
478        >> notify my new successor
479        >> update my table & predecessors table */
480     break;
481
482   case PREDECESSOR_ALIVE:
483     XBT_DEBUG("Receiving a 'Predecessor Alive' request from %s", message->issuer_host_name.c_str());
484     message->type = PREDECESSOR_ALIVE_ANSWER;
485     XBT_DEBUG("Sending back a 'Predecessor Alive Answer' to %s (mailbox %s)",
486         message->issuer_host_name.c_str(), message->answer_to->name());
487     simgrid::s4u::this_actor::dsend(message->answer_to, message, 10);
488     break;
489
490   default:
491     XBT_DEBUG("Ignoring unexpected message: %d from %s", message->type, message->issuer_host_name.c_str());
492     delete message;
493   }
494 }