examples/java/dht/chord/Node.java

   1 /* Copyright (c) 2006-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 package dht.chord;
   7
   8 import org.simgrid.msg.Msg;
   9 import org.simgrid.msg.Comm;
  10 import org.simgrid.msg.Host;
  11 import org.simgrid.msg.Task;
  12 import org.simgrid.msg.Process;
  13 import org.simgrid.msg.MsgException;
  14 import org.simgrid.msg.TimeoutException;
  15 public class Node extends Process {
  16   protected int id;
  17   protected String mailbox;
  18   protected int predId;
  19   protected String predMailbox;
  20   protected int nextFingerToFix;
  21   protected Comm commReceive;
  22   ///Last time I changed a finger or my predecessor
  23   protected double lastChangeDate;
  24   private int[] fingers;
  25
  26   public Node(Host host, String name, String[] args) {
  27     super(host,name,args);
  28   }
  29
  30   @Override
  31   public void main(String[] args) throws MsgException {
  32     if (args.length != 2 && args.length != 4) {
  33       Msg.info("You need to provide 2 or 4 arguments.");
  34       return;
  35     }
  36     double initTime = Msg.getClock();
  37     int i;
  38     boolean joinSuccess;
  39     double deadline;
  40
  41     double nextStabilizeDate = initTime + Common.PERIODIC_STABILIZE_DELAY;
  42     double nextFixFingersDate = initTime + Common.PERIODIC_FIX_FINGERS_DELAY;
  43     double nextCheckPredecessorDate = initTime + Common.PERIODIC_CHECK_PREDECESSOR_DELAY;
  44     double nextLookupDate = initTime + Common.PERIODIC_LOOKUP_DELAY;
  45
  46     mailbox = args[0];
  47     id = Integer.parseInt(args[0]);
  48
  49     fingers = new int[Common.NB_BITS];
  50     for (i = 0; i < Common.NB_BITS; i++) {
  51       fingers[i] = -1;
  52       setFinger(i,this.id);
  53     }
  54
  55     //First node
  56     if (args.length == 2) {
  57       deadline = Integer.parseInt(args[1]);
  58       create();
  59       joinSuccess = true;
  60     } else {
  61       int knownId = Integer.valueOf(args[1]);
  62       deadline = Integer.valueOf(args[3]);
  63       Msg.debug("Hey! Let's join the system with the id " + id + ".");
  64
  65       joinSuccess = join(knownId);
  66     }
  67     if (joinSuccess) {
  68       double currentClock = Msg.getClock();
  69       while (currentClock < (initTime + deadline) && currentClock < Common.MAX_SIMULATION_TIME) {
  70         if (commReceive == null) {
  71           commReceive = Task.irecv(this.mailbox);
  72         }
  73         try {
  74           if (!commReceive.test()) {
  75             if (currentClock >= nextStabilizeDate) {
  76               stabilize();
  77               nextStabilizeDate = Msg.getClock() + Common.PERIODIC_STABILIZE_DELAY;
  78             } else if (currentClock >= nextFixFingersDate) {
  79               fixFingers();
  80               nextFixFingersDate = Msg.getClock() + Common.PERIODIC_FIX_FINGERS_DELAY;
  81             } else if (currentClock >= nextCheckPredecessorDate) {
  82               this.checkPredecessor();
  83               nextCheckPredecessorDate = Msg.getClock() + Common.PERIODIC_CHECK_PREDECESSOR_DELAY;
  84             } else if (currentClock >= nextLookupDate) {
  85               this.randomLookup();
  86               nextLookupDate = Msg.getClock() + Common.PERIODIC_LOOKUP_DELAY;
  87             } else {
  88               waitFor(5);
  89             }
  90             currentClock = Msg.getClock();
  91           } else {
  92             handleTask(commReceive.getTask());
  93             currentClock = Msg.getClock();
  94             commReceive = null;
  95           }
  96         }
  97         catch (Exception e) {
  98           currentClock = Msg.getClock();
  99           commReceive = null;
 100         }
 101       }
 102       leave();
 103       if (commReceive != null) {
 104         commReceive = null;
 105       }
 106     } else {
 107       Msg.info("I couldn't join the ring");
 108     }
 109   }
 110
 111   private void handleTask(Task task) {
 112     if (task instanceof FindSuccessorTask) {
 113       FindSuccessorTask fTask = (FindSuccessorTask)task;
 114       Msg.debug("Receiving a 'Find Successor' request from " + fTask.getIssuerHostName() + " for id " +
 115                 fTask.getRequestId());
 116       // is my successor the successor?
 117       if (isInInterval(fTask.getRequestId(), this.id + 1, fingers[0])) {
 118         Msg.debug("Send the request to " + fTask.getAnswerTo() + " with answer " + fingers[0]);
 119         FindSuccessorAnswerTask answer = new FindSuccessorAnswerTask(getHost().getName(), mailbox, fingers[0]);
 120         answer.dsend(fTask.getAnswerTo());
 121       } else {
 122         // otherwise, forward the request to the closest preceding finger in my table
 123         int closest = closestPrecedingNode(fTask.getRequestId());
 124         Msg.debug("Forward the request to " + closest);
 125         fTask.dsend(Integer.toString(closest));
 126       }
 127     } else if (task instanceof GetPredecessorTask) {
 128       GetPredecessorTask gTask = (GetPredecessorTask)(task);
 129       Msg.debug("Receiving a 'Get Predecessor' request from " + gTask.getIssuerHostName());
 130       GetPredecessorAnswerTask answer = new GetPredecessorAnswerTask(getHost().getName(), mailbox, predId);
 131       answer.dsend(gTask.getAnswerTo());
 132     } else if (task instanceof NotifyTask) {
 133       NotifyTask nTask = (NotifyTask)task;
 134       notify(nTask.getRequestId());
 135     } else {
 136       Msg.debug("Ignoring unexpected task of type:" + task);
 137     }
 138   }
 139
 140   private void leave() {
 141     Msg.debug("Well Guys! I Think it's time for me to quit ;)");
 142     quitNotify(1); //Notify my successor
 143     quitNotify(-1); //Notify my predecessor.
 144   }
 145
 146   /**
 147    * @brief Notifies the successor or the predecessor of the current node of the departure
 148    * @param to 1 to notify the successor, -1 to notify the predecessor
 149    */
 150   static private void quitNotify( int to) {
 151     //TODO
 152   }
 153
 154   /** @brief Initializes the current node as the first one of the system  */
 155   private void create() {
 156     Msg.debug("Create a new Chord ring...");
 157     setPredecessor(-1);
 158   }
 159
 160   // Makes the current node join the ring, knowing the id of a node already in the ring
 161   private boolean join(int knownId) {
 162     Msg.info("Joining the ring with id " + this.id + " knowing node " + knownId);
 163     setPredecessor(-1);
 164     int successorId = remoteFindSuccessor(knownId, this.id);
 165     if (successorId == -1) {
 166       Msg.info("Cannot join the ring.");
 167     } else {
 168       setFinger(0, successorId);
 169     }
 170     return successorId != -1;
 171   }
 172
 173   private void setPredecessor(int predecessorId) {
 174     if (predecessorId != predId) {
 175       predId = predecessorId;
 176       if (predecessorId != -1) {
 177         predMailbox = Integer.toString(predId);
 178       }
 179       lastChangeDate = Msg.getClock();
 180     }
 181   }
 182
 183   /**
 184    * @brief Asks another node its predecessor.
 185    * @param askTo the node to ask to
 186    * @return the id of its predecessor node, or -1 if the request failed(or if the node does not know its predecessor)
 187    */
 188   private int remoteGetPredecessor(int askTo) {
 189     int predecessorId = -1;
 190     boolean stop = false;
 191     Msg.debug("Sending a 'Get Predecessor' request to " + askTo);
 192     String mailboxTo = Integer.toString(askTo);
 193     GetPredecessorTask sendTask = new GetPredecessorTask(getHost().getName(), this.mailbox);
 194     try {
 195       sendTask.send(mailboxTo, Common.TIMEOUT);
 196       try {
 197         do {
 198           if (commReceive == null) {
 199             commReceive = Task.irecv(this.mailbox);
 200           }
 201           commReceive.waitCompletion(Common.TIMEOUT);
 202           Task taskReceived = commReceive.getTask();
 203           if (taskReceived instanceof GetPredecessorAnswerTask) {
 204             predecessorId = ((GetPredecessorAnswerTask) taskReceived).getAnswerId();
 205             stop = true;
 206           } else {
 207             handleTask(taskReceived);
 208           }
 209           commReceive = null;
 210         } while (!stop);
 211       }
 212       catch (MsgException e) {
 213         commReceive = null;
 214         stop = true;
 215       }
 216     }
 217     catch (MsgException e) {
 218       Msg.debug("Failed to send the Get Predecessor request");
 219     }
 220     return predecessorId;
 221   }
 222
 223   /**
 224    * @brief Makes the current node find the successor node of an id.
 225    * @param node the current node
 226    * @param id the id to find
 227    * @return the id of the successor node, or -1 if the request failed
 228    */
 229   private int findSuccessor(int id) {
 230     if (isInInterval(id, this.id + 1, fingers[0])) {
 231       return fingers[0];
 232     }
 233
 234     int closest = this.closestPrecedingNode(id);
 235     return remoteFindSuccessor(closest, id);
 236   }
 237
 238   // Asks another node the successor node of an id.
 239   private int remoteFindSuccessor(int askTo, int id) {
 240     int successor = -1;
 241     boolean stop = false;
 242     String askToMailbox = Integer.toString(askTo);
 243     Task sendTask = new FindSuccessorTask(getHost().getName(), this.mailbox, id);
 244     Msg.debug("Sending a 'Find Successor' request to " + askToMailbox + " for id " + id);
 245     try {
 246       sendTask.send(askToMailbox, Common.TIMEOUT);
 247       do {
 248         if (commReceive == null) {
 249           commReceive = Task.irecv(this.mailbox);
 250         }
 251         try {
 252           commReceive.waitCompletion(Common.TIMEOUT);
 253           Task task = commReceive.getTask();
 254           if (task instanceof FindSuccessorAnswerTask) {
 255             //TODO: Check if this this our answer.
 256             FindSuccessorAnswerTask fTask = (FindSuccessorAnswerTask) task;
 257             stop = true;
 258             successor = fTask.getAnswerId();
 259           } else {
 260             handleTask(task);
 261           }
 262           commReceive = null;
 263         }
 264         catch (TimeoutException e) {
 265           stop = true;
 266           commReceive = null;
 267         }
 268       } while (!stop);
 269     }
 270     catch (TimeoutException e) {
 271       Msg.debug("Failed to send the 'Find Successor' request");
 272     }
 273     catch (MsgException e) {
 274       Msg.debug("Failed to receive Find Successor");
 275     }
 276
 277     return successor;
 278   }
 279
 280   // This function is called periodically. It checks the immediate successor of the current node.
 281   private void stabilize() {
 282     Msg.debug("Stabilizing node");
 283     int candidateId;
 284     int successorId = fingers[0];
 285     if (successorId != this.id){
 286       candidateId = remoteGetPredecessor(successorId);
 287     } else {
 288       candidateId = predId;
 289     }
 290     //This node is a candidate to become my new successor
 291     if (candidateId != -1 && isInInterval(candidateId, this.id + 1, successorId - 1)) {
 292       setFinger(0, candidateId);
 293     }
 294     if (successorId != this.id) {
 295       remoteNotify(successorId, this.id);
 296     }
 297   }
 298
 299   /**
 300    * @brief Notifies the current node that its predecessor may have changed.
 301    * @param candidate_id the possible new predecessor
 302    */
 303   private void notify(int predecessorCandidateId) {
 304     if (predId == -1 || isInInterval(predecessorCandidateId, predId + 1, this.id - 1 )) {
 305       setPredecessor(predecessorCandidateId);
 306     }
 307   }
 308
 309   /**
 310    * @brief Notifies a remote node that its predecessor may have changed.
 311    * @param notify_id id of the node to notify
 312    * @param candidate_id the possible new predecessor
 313    */
 314   private void remoteNotify(int notifyId, int predecessorCandidateId) {
 315     Msg.debug("Sending a 'Notify' request to " + notifyId);
 316     Task sentTask = new NotifyTask(getHost().getName(), this.mailbox, predecessorCandidateId);
 317     sentTask.dsend(Integer.toString(notifyId));
 318   }
 319
 320   // This function is called periodically.
 321   // It refreshes the finger table of the current node.
 322   private void fixFingers() {
 323     Msg.debug("Fixing fingers");
 324     int i = this.nextFingerToFix;
 325     int successorId = this.findSuccessor(this.id + (int)Math.pow(2,i)); //FIXME: SLOW
 326     if (successorId != -1) {
 327       if (successorId != fingers[i]) {
 328         setFinger(i, successorId);
 329       }
 330       nextFingerToFix = (i + 1) % Common.NB_BITS;
 331     }
 332   }
 333
 334   // This function is called periodically.
 335   // It checks whether the predecessor has failed
 336   private void checkPredecessor() {
 337     //TODO
 338   }
 339
 340   // Performs a find successor request to a random id.
 341   private void randomLookup() {
 342     int id = 1337;
 343     //Msg.info("Making a lookup request for id " + id);
 344     findSuccessor(id);
 345   }
 346
 347   /**
 348    * @brief Returns the closest preceding finger of an id with respect to the finger table of the current node.
 349    * @param id the id to find
 350    * @return the closest preceding finger of that id
 351    */
 352   private int closestPrecedingNode(int id) {
 353     for (int i = Common.NB_BITS - 1; i >= 0; i--) {
 354       if (isInInterval(fingers[i], this.id + 1, id - 1)) {
 355         return fingers[i];
 356       }
 357     }
 358     return this.id;
 359   }
 360
 361   /**
 362    * @brief Returns whether an id belongs to the interval [start, end].
 363    *
 364    * The parameters are noramlized to make sure they are between 0 and nb_keys - 1).
 365    * 1 belongs to [62, 3]
 366    * 1 does not belong to [3, 62]
 367    * 63 belongs to [62, 3]
 368    * 63 does not belong to [3, 62]
 369    * 24 belongs to [21, 29]
 370    * 24 does not belong to [29, 21]
 371    *
 372    * @param id id to check
 373    * @param start lower bound
 374    * @param end upper bound
 375    * @return a non-zero value if id in in [start, end]
 376    */
 377   static private boolean isInInterval(int id, int start, int end) {
 378     int normId = normalize(id);
 379     int normStart = normalize(start);
 380     int normEnd = normalize(end);
 381
 382     // make sure end >= start and id >= start
 383     if (normEnd < normStart) {
 384       normEnd += Common.NB_KEYS;
 385     }
 386     if (normId < normStart) {
 387       normId += Common.NB_KEYS;
 388     }
 389     return (normId <= normEnd);
 390   }
 391
 392   /**
 393    * @brief Turns an id into an equivalent id in [0, nb_keys).
 394    * @param id an id
 395    * @return the corresponding normalized id
 396    */
 397   static private int normalize(int id) {
 398     return id & (Common.NB_KEYS - 1);
 399   }
 400
 401   /**
 402    * @brief Sets a finger of the current node.
 403    * @param finger_index index of the finger to set (0 to nb_bits - 1)
 404    * @param id the id to set for this finger
 405    */
 406   private void setFinger(int fingerIndex, int id) {
 407     if (id != fingers[fingerIndex]) {
 408       fingers[fingerIndex] = id;
 409       lastChangeDate = Msg.getClock();
 410     }
 411   }
 412 }