examples/java/kademlia/Node.java

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