examples/lua/chord/chord.lua

   1 -- A SimGrid Lua implementation of the Chord DHT
   2
   3 require("simgrid")
   4
   5 nb_bits = 24
   6 nb_keys = 2^nb_bits
   7 comp_size = 0
   8 comm_size = 10
   9 timeout = 50
  10 max_simulation_time = 1000
  11 stabilize_delay = 20
  12 fix_fingers_delay = 120
  13 check_predecessor_delay = 120
  14 lookup_delay = 10
  15
  16 -- current node (don't worry, globals are duplicated in each process)
  17 my_node = {
  18   id = my_id,
  19   next_finger_to_fix = 1,
  20   fingers = {},
  21   predecessor = nil,
  22   comm_recv = nil
  23 }
  24
  25 -- Main function of each Chord process
  26 -- Arguments:
  27 -- - my id
  28 -- - the id of a guy I know in the system (except for the first node)
  29 function node(...)
  30
  31   -- TODO simplify the parameters
  32   local known_id
  33   local args = {...}
  34   my_node.id = tonumber(args[1])
  35   if #args == 4 then
  36     known_id = tonumber(args[2])
  37   end
  38
  39   simgrid.debug("Hello, I'm a node")
  40
  41   -- join the ring
  42   local join_success = false
  43   if known_id == nil then
  44     -- first node
  45     create()
  46     join_success = true
  47   else
  48     join_success = join(known_id)
  49   end
  50
  51   -- main loop
  52   if join_success then
  53
  54     local now = simgrid.get_clock()
  55     local next_stabilize_date = now + stabilize_delay
  56     local next_fix_fingers_date = now + fix_fingers_delay
  57     local next_check_predecessor_date = now + check_predecessor_delay
  58     local next_lookup_date = now + lookup_delay
  59
  60     local task, success
  61     my_node.comm_recv = simgrid.task.irecv(my_node.id)
  62
  63     while now < max_simulation_time do
  64
  65       task, success = simgrid.comm.test(my_node.comm_recv)
  66
  67       if task then
  68         -- I received a task: answer it
  69         my_node.comm_recv = simgrid.task.irecv(my_node.id)
  70         handle_task(task)
  71       elseif failed then
  72         -- the communication has failed: nevermind
  73         my_node.comm_recv = simgrid.task.irecv(my_node.id)
  74       else
  75         -- no task was received: do periodic calls
  76         if now >= next_stabilize_date then
  77           stabilize()
  78           next_stabilize_date = simgrid.get_clock() + stabilize_delay
  79
  80         elseif now >= next_fix_fingers_date then
  81           fix_fingers()
  82           next_fix_fingers_date = simgrid.get_clock() + fix_fingers_delay
  83
  84         elseif now >= next_check_predecessor_date then
  85           check_predecessor()
  86           next_check_predecessor_date = simgrid.get_clock() + check_predecessor_delay
  87         elseif now >= next_lookup_date then
  88           random_lookup()
  89           next_lookup_date = simgrid.get_clock() + lookup_delay
  90
  91         else
  92           -- nothing to do: sleep for a while
  93           simgrid.process.sleep(5)
  94         end
  95       end
  96       now = simgrid.get_clock()
  97     end -- while
  98
  99     -- leave the ring
 100     leave()
 101   end
 102 end
 103
 104 -- This function is called when the current node receives a task.
 105 -- - task: the task received
 106 function handle_task(task)
 107
 108   local type = task.type
 109
 110   if type == "find successor" then
 111
 112     -- is my successor the successor?
 113     if is_in_interval(task.request_id, my_node.id + 1, my_node.fingers[1]) then
 114       task.type = "find successor answer"
 115       task.answer = my_node.fingers[1]
 116       task:dsend(task.answer_to)
 117     else
 118       -- forward the request to the closest preceding finger in my table
 119       task:dsend(closest_preceding_node(task.request_id))
 120     end
 121
 122   elseif type == "get predecessor" then
 123     task.answer = my_node.predecessor
 124     task:dsend(task.answer_to)
 125
 126   elseif type == "notify" then
 127     -- someone is telling me that he may be my new predecessor
 128     notify(task.request_id)
 129
 130   elseif type == "predecessor leaving" then
 131     -- TODO
 132
 133   elseif type == "successor_leaving" then
 134     -- TODO
 135
 136   elseif type == "find successor answer" then
 137     -- ignoring
 138
 139   elseif type == "get predecessor answer" then
 140     -- ignoring
 141
 142   else
 143     error("Unknown type of task received: " .. task.type)
 144   end
 145 end
 146
 147 -- Returns whether an id belongs to the interval [a, b[.
 148 -- The parameters are normalized to make sure they are between 0 and nb_keys - 1.
 149 -- 1 belongs to [62, 3].
 150 -- 1 does not belong to [3, 62].
 151 -- 63 belongs to [62, 3].
 152 -- 63 does not belong to [3, 62].
 153 -- 24 belongs to [21, 29].
 154 -- 24 does not belong to [29, 21].
 155 function is_in_interval(id, a, b)
 156
 157   -- normalize the parameters
 158   id = id % nb_bits
 159   a = a % nb_bits
 160   b = b % nb_bits
 161
 162   -- make sure a <= b and a <= id
 163   if b < a then
 164     b = b + nb_keys
 165   end
 166
 167   if id < a then
 168     id = id + nb_keys
 169   end
 170
 171   return id <= b
 172 end
 173
 174 -- Creates a new Chord ring.
 175 function create()
 176   my_node.predecessor = nil
 177 end
 178
 179 -- Attemps to join the Chord ring.
 180 -- - known_id: id of a node already in the ring
 181 -- - return value: true if the join was successful
 182 function join(known_id)
 183
 184   local successor = remote_find_successor(known_id, my_node.id)
 185   if successor == nil then
 186     return false
 187   end
 188
 189   my_node.finger[1] = successor
 190   return true
 191 end
 192
 193 -- Returns the closest preceding finger of an id with respect to the finger
 194 -- table of the current node.
 195 -- - id: the id to find
 196 -- - return value: the closest preceding finger of that id
 197 function closest_preceding_node(id)
 198
 199   for i = nb_bits, 1, -1 do
 200     if is_in_interval(my_node.fingers[i].id, my_node.id + 1, id - 1) then
 201       -- finger i is the first one before id
 202       return my_node.fingers[i].id
 203     end
 204   end
 205 end
 206
 207 -- Finds the successor of an id
 208 -- id: the id to find
 209 -- return value: the id of the successor, or -1 if the request failed
 210 function find_successor(id)
 211
 212   if is_in_interval(id, my_node.id + 1, my_node.fingers[1]) then
 213     -- my successor is the successor
 214     return my_node.fingers[1]
 215   end
 216
 217   -- ask to the closest preceding finger in my table
 218   local ask_to = closest_preceding_node(id)
 219   return remote_find_successor(ask_to, id)
 220 end
 221
 222 -- Asks a remote node the successor of an id.
 223 -- ask_to: id of a remote node to ask to
 224 -- id: the id to find
 225 -- return value: the id of the successor, or nil if the request failed
 226 function remote_find_successor(ask_to, id)
 227
 228   local task = simgrid.task.new("", comp_size, comm_size)
 229   task.type = "find successor"
 230   task.request_id = id
 231   task.answer_to = my_node.id
 232
 233   if task:send(ask_to, timeout) then
 234     -- request successfully sent: wait for an answer
 235     while true do
 236       task = simgrid.comm.wait(my_node.comm_recv, timeout)
 237       my_node.comm_recv = simgrid.task.irecv(my_node.id)
 238
 239       if not task then
 240         -- failed to receive the answer
 241         return nil
 242       else
 243         -- a task was received: is it the expected answer?
 244         if task.type ~= "find successor answer" or task[request_id] ~= id then
 245           -- this is not our answer
 246           handle_task(task)
 247         else
 248           -- this is our answer
 249           return task.answer
 250         end
 251       end
 252     end
 253   end
 254
 255   return successor
 256 end
 257
 258 -- Asks a remote node its predecessor.
 259 -- ask_to: id of a remote node to ask to
 260 -- return value: the id of its predecessor, or nil if the request failed
 261 function remote_get_predecessor(ask_to)
 262
 263   local task = simgrid.task.new("", comp_size, comm_size)
 264   task.type = "get predecessor"
 265   task.answer_to = my_node.id
 266
 267   if task:send(ask_to, timeout) then
 268     -- request successfully sent: wait for an answer
 269     while true do
 270       task = simgrid.comm.wait(my_node.comm_recv, timeout)
 271       my_node.comm_recv = simgrid.task.irecv(my_node.id)
 272
 273       if not task then
 274         -- failed to receive the answer
 275         return nil
 276       else
 277         -- a task was received: is it the expected answer?
 278         if task.type ~= "get predecessor answer" then
 279           -- this is not our answer
 280           handle_task(task)
 281         else
 282           -- this is our answer
 283           -- FIXME make sure the message answers to this particular request
 284           return task.answer
 285         end
 286       end
 287     end
 288   end
 289
 290   return successor
 291 end
 292
 293 -- Checks the immediate successor of the current node.
 294 function stabilize()
 295
 296   local candidate
 297   local successor = my_node.fingers[1]
 298   if successor ~= my_node.id then
 299     candidate = remote_get_predecessor(successor)
 300   else
 301     candidate = my_node.predecessor
 302   end
 303
 304   -- this node is a candidate to become my new successor
 305   if candidate ~= nil and is_in_interval(candidate, my_node.id + 1, successor -1) then
 306     my_node.fingers[1] = candidate
 307   end
 308
 309   if successor ~= my_node.id then
 310     remote_notify(successor, my_node.id)
 311   end
 312 end
 313
 314 -- Notifies the current node that its predecessor my have changed
 315 -- - candidate_predecessor: the possible new predecessor
 316 function notify(candidate_predecessor)
 317
 318   if my_node.predecessor == nil or is_in_interval(candidate_predecessor,
 319       my_node.predecessor + 1, my_node.id - 1) then
 320     my_node.predecessor = candidate_predecessor
 321   end
 322 end
 323
 324 -- Notifies a remote node that its predecessor my have changed.
 325 -- - notify_to
 326 -- - candidate the possible new predecessor
 327 function remote_notify(notify_to, candidate_predecessor)
 328
 329   local task = simgrid.task.new("", comp_size, comm_size)
 330   task.type = "notify"
 331   task.request_id = candidate_predecessor
 332   task:dsend(notify_to)
 333 end
 334
 335 -- Refreshes the finger table of the current node.
 336 function fix_fingers()
 337
 338   local i = my_node.next_finger_to_fix
 339   local id = find_successor(my_node.id + 2^i)
 340   if id ~= nil then
 341     if id ~= my_node.fingers[i] then
 342       fingers[i] = id
 343     end
 344     my_node.next_finger_to_fix = (i % nb_bits) + 1
 345   end
 346 end
 347
 348 -- Checks whether the predecessor of the current node has failed.
 349 function check_predecessor()
 350   -- TODO
 351 end
 352
 353 -- Performs a find successor request to an arbitrary id.
 354 function random_lookup()
 355
 356   find_successor(1337)
 357 end
 358
 359 simgrid.platform(arg[1] or "../../msg/msg_platform.xml")
 360 simgrid.application(arg[2] or "../../msg/chord/chord90.xml")
 361 simgrid.run()
 362