examples/java/dht/kademlia/Node.java

   1 /* Copyright (c) 2012-2019. The SimGrid Team.
   2  * All rights reserved.                                                     */
   3
   4 /* This program is free software; you can redistribute it and/or modify it
   5  * under the terms of the license (GNU LGPL) which comes with this package. */
   6
   7 package dht.kademlia;
   8
   9 import org.simgrid.msg.Host;
  10
  11 import org.simgrid.msg.Msg;
  12 import org.simgrid.msg.Comm;
  13 import org.simgrid.msg.Task;
  14 import org.simgrid.msg.Process;
  15 import org.simgrid.msg.MsgException;
  16
  17 public class Node extends Process {
  18   protected int id;
  19   protected RoutingTable table;
  20   protected int deadline;
  21   protected int findNodeSuccedded = 0;
  22   protected int findNodeFailed = 0;
  23   protected Comm comm;
  24
  25   public Node(Host host, String name, String[]args) {
  26     super(host,name,args);
  27   }
  28
  29   @Override
  30   public void main(String[] args) throws MsgException {
  31     //Check the number of arguments.
  32     if (args.length != 2 && args.length != 3) {
  33       Msg.info("Wrong argument count.");
  34       return;
  35     }
  36     this.id = Integer.parseInt(args[0]);
  37     this.table = new RoutingTable(this.id);
  38
  39     if (args.length == 3) {
  40       this.deadline = Integer.parseInt(args[2]);
  41       Msg.info("Hi, I'm going to join the network with the id " + id + "!");
  42       if (joinNetwork(Integer.parseInt(args[1]))) {
  43         this.mainLoop();
  44       }
  45       else {
  46         Msg.info("I couldn't join the network :(");
  47       }
  48     }
  49     else {
  50       this.deadline = Integer.parseInt(args[1]);
  51       Msg.info("Hi, I'm going to create the network with the id " + id + "!");
  52       table.update(this.id);
  53       this.mainLoop();
  54     }
  55     Msg.debug("I'm leaving the network");
  56     Msg.debug("Here is my routing table:" + table);
  57   }
  58
  59   public void mainLoop() {
  60     double nextLookupTime = Msg.getClock() + Common.RANDOM_LOOKUP_INTERVAL;
  61     while (Msg.getClock() < this.deadline) {
  62       try {
  63         if (comm == null) {
  64           comm = Task.irecv(Integer.toString(id));
  65         }
  66         if (!comm.test()) {
  67           if (Msg.getClock() >= nextLookupTime) {
  68             randomLookup();
  69             nextLookupTime += Common.RANDOM_LOOKUP_INTERVAL;
  70           } else {
  71             waitFor(1);
  72           }
  73         } else {
  74           Task task = comm.getTask();
  75           handleTask(task);
  76           comm = null;
  77         }
  78       }
  79       catch (Exception e) {
  80         Msg.debug("Caught exception: " + e);
  81       }
  82     }
  83     Msg.info(findNodeSuccedded + "/"  + (findNodeSuccedded + findNodeFailed) + " FIND_NODE have succedded.");
  84   }
  85
  86   /**
  87    * @brief Try to make the node join the network
  88    * @param idKnown Id of someone we know in the system
  89    */
  90   public boolean joinNetwork(int idKnown) {
  91     boolean answerGot = false;
  92     double timeBegin = Msg.getClock();
  93     Msg.debug("Joining the network knowing " + idKnown);
  94     //Add ourselves and the node we know to our routing table
  95     table.update(this.id);
  96     table.update(idKnown);
  97     //Send a "FIND_NODE" to the node we know.
  98     sendFindNode(idKnown,this.id);
  99     //Wait for the answer.
 100     int trials = 0;
 101
 102     do {
 103       try {
 104         if (comm == null) {
 105           comm = Task.irecv(Integer.toString(id));
 106         }
 107         if (comm != null) {
 108           if (!comm.test()) {
 109             waitFor(1);
 110           } else {
 111             Task task = comm.getTask();
 112             if (task instanceof FindNodeAnswerTask) {
 113               //Retrieve the node list and ping them
 114               FindNodeAnswerTask answerTask = (FindNodeAnswerTask)task;
 115               Answer answer = answerTask.getAnswer();
 116               answerGot = true;
 117               if (answer.getDestinationId() == this.id) {
 118                 //Ping everyone in the list
 119                 for (Contact c : answer.getNodes()) {
 120                   table.update(c.getId());
 121                 }
 122               }
 123             } else {
 124               handleTask(task);
 125             }
 126             comm = null;
 127           }
 128         }
 129       }
 130       catch (Exception ex) {
 131         trials++;
 132         Msg.info("FIND_NODE failed");
 133       }
 134     } while (!answerGot && trials < Common.MAX_JOIN_TRIALS);
 135     /* Second step: Send a FIND_NODE in a node in each bucket */
 136     int bucketId = table.findBucket(idKnown).getId();
 137     for (int i = 0; ((bucketId - i) > 0 ||
 138        (bucketId + i) <= Common.IDENTIFIER_SIZE) &&
 139        i < Common.JOIN_BUCKETS_QUERIES; i++) {
 140       if (bucketId - i > 0) {
 141         int idInBucket = table.getIdInPrefix(this.id,bucketId - i);
 142         this.findNode(idInBucket,false);
 143       }
 144       if (bucketId + i <= Common.IDENTIFIER_SIZE) {
 145         int idInBucket = table.getIdInPrefix(this.id,bucketId + i);
 146         findNode(idInBucket,false);
 147       }
 148     }
 149     Msg.debug("Time spent:" + (Msg.getClock() - timeBegin));
 150     return answerGot;
 151   }
 152
 153   /* Send a request to find a node in the node's routing table. */
 154   public boolean findNode(int destination, boolean counts) {
 155     int queries;
 156     int answers;
 157     int nodesAdded;
 158     boolean destinationFound;
 159     int steps = 0;
 160     double timeBeginReceive;
 161     double timeout;
 162     double globalTimeout = Msg.getClock() + Common.FIND_NODE_GLOBAL_TIMEOUT;
 163     //Build a list of the closest nodes we already know.
 164     Answer nodeList = table.findClosest(destination);
 165     Msg.verb("Doing a FIND_NODE on " + destination);
 166     do {
 167       timeBeginReceive = Msg.getClock();
 168       answers = 0;
 169       queries = this.sendFindNodeToBest(nodeList);
 170       nodesAdded = 0;
 171       timeout = Msg.getClock() + Common.FIND_NODE_TIMEOUT;
 172       steps++;
 173       do {
 174         try {
 175           if (comm == null) {
 176             comm = Task.irecv(Integer.toString(id));
 177           }
 178           if (!comm.test()) {
 179             waitFor(1);
 180           } else {
 181             Task task = comm.getTask();
 182             if (task instanceof FindNodeAnswerTask) {
 183               FindNodeAnswerTask answerTask = (FindNodeAnswerTask)task;
 184               //Check if we received what we are looking for.
 185               if (answerTask.getDestinationId() == destination) {
 186                 table.update(answerTask.getSenderId());
 187                 //Add the answer to our routing table
 188                 for (Contact c: answerTask.getAnswer().getNodes()) {
 189                   table.update(c.getId());
 190                 }
 191                 answers++;
 192
 193                 nodesAdded = nodeList.merge(answerTask.getAnswer());
 194               } else {
 195               /* If it's not our answer, we answer to the node that has queried us anyway */
 196                 handleTask(task);
 197                 //Update the timeout if it's not our answer.
 198                 timeout += Msg.getClock() - timeBeginReceive;
 199                 timeBeginReceive = Msg.getClock();
 200               }
 201             } else {
 202               handleTask(task);
 203               timeout += Msg.getClock() - timeBeginReceive;
 204               timeBeginReceive = Msg.getClock();
 205             }
 206             comm = null;
 207           }
 208         }
 209         catch (Exception e) {
 210           comm = null;
 211         }
 212       } while (answers < queries && Msg.getClock() < timeout);
 213       destinationFound = nodeList.destinationFound();
 214     } while (!destinationFound && (nodesAdded > 0 || answers == 0) && Msg.getClock() < globalTimeout
 215              && steps < Common.MAX_STEPS);
 216
 217     if (destinationFound) {
 218       if (counts) {
 219         findNodeSuccedded++;
 220       }
 221       Msg.debug("Find node on " + destination + " succedded");
 222     } else {
 223       Msg.debug("Find node on " + destination + " failed");
 224       Msg.debug("Queried " + queries + " nodes to find "  + destination);
 225       Msg.debug(nodeList.toString());
 226       if (counts) {
 227         findNodeFailed++;
 228       }
 229     }
 230     return destinationFound;
 231   }
 232
 233   /**
 234    * @brief Sends a "PING" request to a node
 235    * @param destination Ping destination id.
 236    */
 237   public void ping(int destination) {
 238     boolean destinationFound = false;
 239     double timeout = Msg.getClock() + Common.PING_TIMEOUT;
 240     PingTask pingTask = new PingTask(this.id);
 241     /* Sending the ping task */
 242     pingTask.dsend(Integer.toString(destination));
 243     do {
 244       try {
 245         Task task = Task.receive(Integer.toString(this.id),Common.PING_TIMEOUT);
 246         if (task instanceof PingAnswerTask) {
 247           PingAnswerTask answerTask = (PingAnswerTask)task;
 248           if (answerTask.getSenderId() == destination) {
 249             this.table.update(destination);
 250             destinationFound = true;
 251           } else {
 252             handleTask(task);
 253           }
 254         } else {
 255           handleTask(task);
 256         }
 257         waitFor(1);
 258       }
 259       catch (Exception ex) {
 260         Msg.debug("Caught exception: " + ex);
 261       }
 262     } while (Msg.getClock() < timeout && !destinationFound);
 263   }
 264
 265   /**
 266    * @brief Sends a "FIND_NODE" request (task) to a node we know.
 267    * @param id Id of the node we are querying
 268    * @param destination id of the node we are trying to find.
 269    */
 270   public void sendFindNode(int id, int destination) {
 271     Msg.debug("Sending a FIND_NODE to " + Integer.toString(id) + " to find " + destination  );
 272     FindNodeTask task = new FindNodeTask(this.id,destination);
 273     task.dsend(Integer.toString(id));
 274   }
 275
 276   /** Sends a "FIND_NODE" request to the best "alpha" nodes in a node list */
 277   public int sendFindNodeToBest(Answer nodeList) {
 278     int destination = nodeList.getDestinationId();
 279     int i;
 280     for (i = 0; i < Common.ALPHA && i < nodeList.size(); i++) {
 281       Contact node = nodeList.getNodes().get(i);
 282       if (node.getId() != this.id) {
 283         this.sendFindNode(node.getId(),destination);
 284       }
 285     }
 286     return i;
 287   }
 288
 289   public void randomLookup() {
 290     findNode(0,true);
 291   }
 292
 293   /**
 294    * @brief Handles an incomming task
 295    * @param task The task we need to handle
 296    */
 297   public void handleTask(Task task) {
 298     if (task instanceof KademliaTask) {
 299       table.update(((KademliaTask) task).getSenderId());
 300       if (task instanceof FindNodeTask) {
 301         handleFindNode((FindNodeTask)task);
 302       }
 303       else if (task instanceof PingTask) {
 304         handlePing((PingTask)task);
 305       }
 306     }
 307   }
 308
 309   public void handleFindNode(FindNodeTask task) {
 310     Msg.debug("Received a FIND_NODE from " + task.getSenderId());
 311     Answer answer = table.findClosest(task.getDestination());
 312     FindNodeAnswerTask taskToSend = new FindNodeAnswerTask(this.id,task.getDestination(),answer);
 313     taskToSend.dsend(Integer.toString(task.getSenderId()));
 314   }
 315
 316   public void handlePing(PingTask task) {
 317     Msg.debug("Received a PING from " + task.getSenderId());
 318     PingAnswerTask taskToSend = new PingAnswerTask(this.id);
 319     taskToSend.dsend(Integer.toString(task.getSenderId()));
 320   }
 321 }