examples/s4u/dht-chord/node.cpp

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