examples/java/dht/kademlia/Node.java

   1 /* Copyright (c) 2012-2014, 2016. 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 next_lookup_time = 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() >= next_lookup_time) {
  68             randomLookup();
  69             next_lookup_time += 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       }
  81     }
  82     Msg.info(findNodeSuccedded + "/"  + (findNodeSuccedded + findNodeFailed) + " FIND_NODE have succedded.");
  83   }
  84
  85   /**
  86    * @brief Try to make the node join the network
  87    * @param idKnown Id of someone we know in the system
  88    */
  89   public boolean joinNetwork(int idKnown) {
  90     boolean answerGot = false;
  91     double timeBegin = Msg.getClock();
  92     Msg.debug("Joining the network knowing " + idKnown);
  93     //Add ourselves and the node we know to our routing table
  94     table.update(this.id);
  95     table.update(idKnown);
  96     //Send a "FIND_NODE" to the node we know.
  97     sendFindNode(idKnown,this.id);
  98     //Wait for the answer.
  99     int trials = 0;
 100
 101     do {
 102       try {
 103         if (comm == null) {
 104           comm = Task.irecv(Integer.toString(id));
 105         }
 106         if (comm != null) {
 107           if (!comm.test()) {
 108             waitFor(1);
 109           } else {
 110             Task task = comm.getTask();
 111             if (task instanceof FindNodeAnswerTask) {
 112               //Retrieve the node list and ping them
 113               FindNodeAnswerTask answerTask = (FindNodeAnswerTask)task;
 114               Answer answer = answerTask.getAnswer();
 115               answerGot = true;
 116               if (answer.getDestinationId() == this.id) {
 117                 //Ping everyone in the list
 118                 for (Contact c : answer.getNodes()) {
 119                   table.update(c.getId());
 120                 }
 121               }
 122             } else {
 123               handleTask(task);
 124             }
 125             comm = null;
 126           }
 127         }
 128       }
 129       catch (Exception ex) {
 130         trials++;
 131         Msg.info("FIND_NODE failed");
 132       }
 133     } while (!answerGot && trials < Common.MAX_JOIN_TRIALS);
 134     /* Second step: Send a FIND_NODE in a node in each bucket */
 135     int bucketId = table.findBucket(idKnown).getId();
 136     for (int i = 0; ((bucketId - i) > 0 ||
 137        (bucketId + i) <= Common.IDENTIFIER_SIZE) &&
 138        i < Common.JOIN_BUCKETS_QUERIES; i++) {
 139       if (bucketId - i > 0) {
 140         int idInBucket = table.getIdInPrefix(this.id,bucketId - i);
 141         this.findNode(idInBucket,false);
 142       }
 143       if (bucketId + i <= Common.IDENTIFIER_SIZE) {
 144         int idInBucket = table.getIdInPrefix(this.id,bucketId + i);
 145         findNode(idInBucket,false);
 146       }
 147     }
 148     Msg.debug("Time spent:" + (Msg.getClock() - timeBegin));
 149     return answerGot;
 150   }
 151
 152   /* Send a request to find a node in the node's routing table. */
 153   public boolean findNode(int destination, boolean counts) {
 154     int queries;
 155     int answers;
 156     int nodesAdded;
 157     boolean destinationFound;
 158     int steps = 0;
 159     double timeBeginReceive;
 160     double timeout;
 161     double globalTimeout = Msg.getClock() + Common.FIND_NODE_GLOBAL_TIMEOUT;
 162     //Build a list of the closest nodes we already know.
 163     Answer nodeList = table.findClosest(destination);
 164     Msg.verb("Doing a FIND_NODE on " + destination);
 165     do {
 166       timeBeginReceive = Msg.getClock();
 167       answers = 0;
 168       queries = this.sendFindNodeToBest(nodeList);
 169       nodesAdded = 0;
 170       timeout = Msg.getClock() + Common.FIND_NODE_TIMEOUT;
 171       steps++;
 172       do {
 173         try {
 174           if (comm == null) {
 175             comm = Task.irecv(Integer.toString(id));
 176           }
 177           if (!comm.test()) {
 178             waitFor(1);
 179           } else {
 180             Task task = comm.getTask();
 181             if (task instanceof FindNodeAnswerTask) {
 182               FindNodeAnswerTask answerTask = (FindNodeAnswerTask)task;
 183               //Check if we received what we are looking for.
 184               if (answerTask.getDestinationId() == destination) {
 185                 table.update(answerTask.getSenderId());
 186                 //Add the answer to our routing table
 187                 for (Contact c: answerTask.getAnswer().getNodes()) {
 188                   table.update(c.getId());
 189                 }
 190                 answers++;
 191
 192                 nodesAdded = nodeList.merge(answerTask.getAnswer());
 193               } else {
 194               /* If it's not our answer, we answer to the node that has queried us anyway */
 195                 handleTask(task);
 196                 //Update the timeout if it's not our answer.
 197                 timeout += Msg.getClock() - timeBeginReceive;
 198                 timeBeginReceive = Msg.getClock();
 199               }
 200             } else {
 201               handleTask(task);
 202               timeout += Msg.getClock() - timeBeginReceive;
 203               timeBeginReceive = Msg.getClock();
 204             }
 205             comm = null;
 206           }
 207         }
 208         catch (Exception e) {
 209           comm = null;
 210         }
 211       } while (answers < queries && Msg.getClock() < timeout);
 212       destinationFound = nodeList.destinationFound();
 213     } while (!destinationFound && (nodesAdded > 0 || answers == 0) && Msg.getClock() < globalTimeout
 214              && steps < Common.MAX_STEPS);
 215
 216     if (destinationFound) {
 217       if (counts) {
 218         findNodeSuccedded++;
 219       }
 220       Msg.debug("Find node on " + destination + " succedded");
 221     } else {
 222       Msg.debug("Find node on " + destination + " failed");
 223       Msg.debug("Queried " + queries + " nodes to find "  + destination);
 224       Msg.debug(nodeList.toString());
 225       if (counts) {
 226         findNodeFailed++;
 227       }
 228     }
 229     return destinationFound;
 230   }
 231
 232   /**
 233    * @brief Sends a "PING" request to a node
 234    * @param destination Ping destination id.
 235    */
 236   public void ping(int destination) {
 237     boolean destinationFound = false;
 238     double timeout = Msg.getClock() + Common.PING_TIMEOUT;
 239     PingTask pingTask = new PingTask(this.id);
 240     /* Sending the ping task */
 241     pingTask.dsend(Integer.toString(destination));
 242     do {
 243       try {
 244         Task task = Task.receive(Integer.toString(this.id),Common.PING_TIMEOUT);
 245         if (task instanceof PingAnswerTask) {
 246           PingAnswerTask answerTask = (PingAnswerTask)task;
 247           if (answerTask.getSenderId() == destination) {
 248             this.table.update(destination);
 249             destinationFound = true;
 250           } else {
 251             handleTask(task);
 252           }
 253         } else {
 254           handleTask(task);
 255         }
 256         waitFor(1);
 257       }
 258       catch (Exception ex) {
 259       }
 260     } while (Msg.getClock() < timeout && !destinationFound);
 261   }
 262
 263   /**
 264    * @brief Sends a "FIND_NODE" request (task) to a node we know.
 265    * @param id Id of the node we are querying
 266    * @param destination id of the node we are trying to find.
 267    */
 268   public void sendFindNode(int id, int destination) {
 269     Msg.debug("Sending a FIND_NODE to " + Integer.toString(id) + " to find " + destination  );
 270     FindNodeTask task = new FindNodeTask(this.id,destination);
 271     task.dsend(Integer.toString(id));
 272   }
 273
 274   /** Sends a "FIND_NODE" request to the best "alpha" nodes in a node list */
 275   public int sendFindNodeToBest(Answer nodeList) {
 276     int destination = nodeList.getDestinationId();
 277     int i;
 278     for (i = 0; i < Common.alpha && i < nodeList.size(); i++) {
 279       Contact node = nodeList.getNodes().get(i);
 280       if (node.getId() != this.id) {
 281         this.sendFindNode(node.getId(),destination);
 282       }
 283     }
 284     return i;
 285   }
 286
 287   public void randomLookup() {
 288     findNode(0,true);
 289   }
 290
 291   /**
 292    * @brief Handles an incomming task
 293    * @param task The task we need to handle
 294    */
 295   public void handleTask(Task task) {
 296     if (task instanceof KademliaTask) {
 297       table.update(((KademliaTask) task).getSenderId());
 298       if (task instanceof FindNodeTask) {
 299         handleFindNode((FindNodeTask)task);
 300       }
 301       else if (task instanceof PingTask) {
 302         handlePing((PingTask)task);
 303       }
 304     }
 305   }
 306
 307   public void handleFindNode(FindNodeTask task) {
 308     Msg.debug("Received a FIND_NODE from " + task.getSenderId());
 309     Answer answer = table.findClosest(task.getDestination());
 310     FindNodeAnswerTask taskToSend = new FindNodeAnswerTask(this.id,task.getDestination(),answer);
 311     taskToSend.dsend(Integer.toString(task.getSenderId()));
 312   }
 313
 314   public void handlePing(PingTask task) {
 315     Msg.debug("Received a PING from " + task.getSenderId());
 316     PingAnswerTask taskToSend = new PingAnswerTask(this.id);
 317     taskToSend.dsend(Integer.toString(task.getSenderId()));
 318   }
 319 }