Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
Add java-bittorrent example (still not working).
[simgrid.git] / 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 simulated 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 deployment file
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   -- initialize the node
40   for i = 1, nb_bits do
41     my_node.fingers[i] = my_node.id
42   end
43   my_node.comm_recv = simgrid.task.irecv(my_node.id)
44
45   -- join the ring
46   local join_success = false
47   if known_id == nil then
48     -- first node
49     create()
50     join_success = true
51   else
52     join_success = join(known_id)
53   end
54
55   -- main loop
56   if join_success then
57
58     local now = simgrid.get_clock()
59     local next_stabilize_date = now + stabilize_delay
60     local next_fix_fingers_date = now + fix_fingers_delay
61     local next_check_predecessor_date = now + check_predecessor_delay
62     local next_lookup_date = now + lookup_delay
63
64     local task, err
65
66     while now < max_simulation_time do
67
68       task, err = my_node.comm_recv:test()
69
70       if task then
71         -- I received a task: answer it
72         my_node.comm_recv = simgrid.task.irecv(my_node.id)
73         handle_task(task)
74       elseif err then
75         -- the communication has failed: nevermind
76         my_node.comm_recv = simgrid.task.irecv(my_node.id)
77       else
78         -- no task was received: do periodic calls
79         if now >= next_stabilize_date then
80           stabilize()
81           next_stabilize_date = simgrid.get_clock() + stabilize_delay
82
83         elseif now >= next_fix_fingers_date then
84           fix_fingers()
85           next_fix_fingers_date = simgrid.get_clock() + fix_fingers_delay
86
87         elseif now >= next_check_predecessor_date then
88           check_predecessor()
89           next_check_predecessor_date = simgrid.get_clock() + check_predecessor_delay
90
91         elseif now >= next_lookup_date then
92           random_lookup()
93           next_lookup_date = simgrid.get_clock() + lookup_delay
94
95         else
96           -- nothing to do: sleep for a while
97           simgrid.process.sleep(5)
98         end
99       end
100       now = simgrid.get_clock()
101     end -- while
102
103     -- leave the ring
104     leave()
105   end
106 end
107
108 -- Makes the current node leave the ring
109 function leave()
110
111   simgrid.info("Leaving the ring")
112   -- TODO: notify others
113 end
114
115 -- This function is called when the current node receives a task.
116 -- - task: the task received
117 function handle_task(task)
118
119   local type = task.type
120
121   if type == "find successor" then
122
123     simgrid.info("Received a 'find successor' request from " .. task.answer_to ..
124         " for id " .. task.request_id)
125
126     -- is my successor the successor?
127     if is_in_interval(task.request_id, my_node.id + 1, my_node.fingers[1]) then
128
129       simgrid.info("Sending back a 'find successor answer' to " ..
130           task.answer_to .. ": the successor of " .. task.request_id ..
131           " is " .. my_node.fingers[1])
132
133       task.type = "find successor answer"
134       task.answer = my_node.fingers[1]
135       task:dsend(task.answer_to)
136     else
137       -- forward the request to the closest preceding finger in my table
138
139       simgrid.info("Forwarding the 'find successor' request to my closest preceding finger")
140       task:dsend(closest_preceding_node(task.request_id))
141     end
142
143   elseif type == "get predecessor" then
144     task.type = "get predecessor answer"
145     task.answer = my_node.predecessor
146     task:dsend(task.answer_to)
147
148   elseif type == "notify" then
149     -- someone is telling me that he may be my new predecessor
150     notify(task.request_id)
151
152   elseif type == "predecessor leaving" then
153     -- TODO
154
155   elseif type == "successor_leaving" then
156     -- TODO
157
158   elseif type == "find successor answer" then
159     -- ignoring
160
161   elseif type == "get predecessor answer" then
162     -- ignoring
163
164   else
165     error("Unknown type of task received: " .. task.type)
166   end
167 end
168
169 -- Returns whether an id belongs to the interval [a, b[.
170 -- The parameters are normalized to make sure they are between 0 and nb_keys - 1.
171 -- 1 belongs to [62, 3].
172 -- 1 does not belong to [3, 62].
173 -- 63 belongs to [62, 3].
174 -- 63 does not belong to [3, 62].
175 -- 24 belongs to [21, 29].
176 -- 24 does not belong to [29, 21].
177 function is_in_interval(id, a, b)
178
179   -- normalize the parameters
180   id = id % nb_bits
181   a = a % nb_bits
182   b = b % nb_bits
183  
184   -- make sure a <= b and a <= id
185   if b < a then
186     b = b + nb_keys
187   end
188
189   if id < a then
190     id = id + nb_keys
191   end
192
193   return id <= b
194 end
195
196 -- Returns whether the current node is in the ring.
197 function has_joined()
198
199   return my_node.fingers[1] ~= nil
200 end
201
202 -- Creates a new Chord ring.
203 function create()
204   my_node.predecessor = nil
205   my_node.fingers[1] = my_node.id
206 end
207
208 -- Attemps to join the Chord ring.
209 -- - known_id: id of a node already in the ring
210 -- - return value: true if the join was successful
211 function join(known_id)
212
213   simgrid.info("Joining the ring with id " .. my_node.id .. ", knowing node " .. known_id)
214
215   local successor = remote_find_successor(known_id, my_node.id)
216   if successor == nil then
217     simgrid.info("Cannot join the ring.")
218     return false
219   end
220
221   my_node.predecessor = nil
222   my_node.fingers[1] = successor
223   return true
224 end
225
226 -- Returns the closest preceding finger of an id with respect to the finger
227 -- table of the current node.
228 -- - id: the id to find
229 -- - return value: the closest preceding finger of that id
230 function closest_preceding_node(id)
231
232   for i = nb_bits, 1, -1 do
233     if is_in_interval(my_node.fingers[i], my_node.id + 1, id - 1) then
234       -- finger i is the first one before id
235       return my_node.fingers[i]
236     end
237   end
238 end
239
240 -- Finds the successor of an id
241 -- id: the id to find
242 -- return value: the id of the successor, or nil if the request failed
243 function find_successor(id)
244
245   if is_in_interval(id, my_node.id + 1, my_node.fingers[1]) then
246     -- my successor is the successor
247     return my_node.fingers[1]
248   end
249
250   -- ask to the closest preceding finger in my table
251   local ask_to = closest_preceding_node(id)
252   return remote_find_successor(ask_to, id)
253 end
254
255 -- Asks a remote node the successor of an id.
256 -- ask_to: id of a remote node to ask to
257 -- id: the id to find
258 -- return value: the id of the successor, or nil if the request failed
259 function remote_find_successor(ask_to, id)
260
261   local task = simgrid.task.new("", comp_size, comm_size)
262   task.type = "find successor"
263   task.request_id = id
264   task.answer_to = my_node.id
265
266   simgrid.info("Sending a 'find successor' request to " .. ask_to .. " for id " .. id)
267   if task:send(ask_to, timeout) then
268     -- request successfully sent: wait for an answer
269
270     simgrid.info("Sent the 'find successor' request to " .. ask_to ..
271         " for id " .. id .. ", waiting for the answer")
272
273     while true do
274       task = my_node.comm_recv:wait(timeout)
275       my_node.comm_recv = simgrid.task.irecv(my_node.id)
276     
277       if not task then
278         -- failed to receive the answer
279         simgrid.info("Failed to receive the answer to my 'find successor' request")
280         return nil
281       else
282         -- a task was received: is it the expected answer?
283         if task.type ~= "find successor answer" or task.request_id ~= id then
284           -- this is not our answer
285           simgrid.info("Received another request of type " .. task.type)
286           handle_task(task)
287         else
288           -- this is our answer
289           simgrid.info("Received the answer to my 'find successor' request for id " ..
290               id .. ": the successor is " .. task.answer)
291           return task.answer
292         end
293       end
294     end
295   else
296     simgrid.info("Failed to send the 'find successor' request to " .. ask_to ..
297         " for id " .. id)
298   end
299
300   return successor
301 end
302
303 -- Asks a remote node its predecessor.
304 -- ask_to: id of a remote node to ask to
305 -- return value: the id of its predecessor, or nil if the request failed
306 function remote_get_predecessor(ask_to)
307
308   local task = simgrid.task.new("", comp_size, comm_size)
309   task.type = "get predecessor"
310   task.answer_to = my_node.id
311
312   if task:send(ask_to, timeout) then
313     -- request successfully sent: wait for an answer
314     while true do
315       task = my_node.comm_recv:wait(timeout)
316       my_node.comm_recv = simgrid.task.irecv(my_node.id)
317     
318       if not task then
319         -- failed to receive the answer
320         return nil
321       else
322         -- a task was received: is it the expected answer?
323         if task.type ~= "get predecessor answer" then
324           -- this is not our answer
325           handle_task(task)
326         else
327           -- this is our answer
328           -- FIXME make sure the message answers to this particular request
329           return task.answer
330         end
331       end
332     end
333   end
334
335   return successor
336 end
337
338 -- Checks the immediate successor of the current node.
339 function stabilize()
340
341   local candidate
342   local successor = my_node.fingers[1]
343   if successor ~= my_node.id then
344     candidate = remote_get_predecessor(successor)
345   else
346     candidate = my_node.predecessor
347   end
348
349   -- this node is a candidate to become my new successor
350   if candidate ~= nil and is_in_interval(candidate, my_node.id + 1, successor - 1) then
351     my_node.fingers[1] = candidate
352   end
353
354   if successor ~= my_node.id then
355     remote_notify(successor, my_node.id)
356   end
357 end
358
359 -- Notifies the current node that its predecessor my have changed
360 -- - candidate_predecessor: the possible new predecessor
361 function notify(candidate_predecessor)
362
363   if my_node.predecessor == nil or is_in_interval(candidate_predecessor,
364       my_node.predecessor + 1, my_node.id - 1) then
365     my_node.predecessor = candidate_predecessor
366   end
367 end
368
369 -- Notifies a remote node that its predecessor my have changed.
370 -- - notify_to
371 -- - candidate the possible new predecessor
372 function remote_notify(notify_to, candidate_predecessor)
373
374   local task = simgrid.task.new("", comp_size, comm_size)
375   task.type = "notify"
376   task.request_id = candidate_predecessor
377   task:dsend(notify_to)
378 end
379
380 -- Refreshes the finger table of the current node.
381 function fix_fingers()
382
383   local i = my_node.next_finger_to_fix
384   local id = find_successor(my_node.id + 2^i)
385   if id ~= nil then
386     if id ~= my_node.fingers[i] then
387       my_node.fingers[i] = id
388     end
389     my_node.next_finger_to_fix = (i % nb_bits) + 1
390   end
391 end
392
393 -- Checks whether the predecessor of the current node has failed.
394 function check_predecessor()
395   -- TODO
396 end
397
398 -- Performs a find successor request to an arbitrary id.
399 function random_lookup()
400
401   find_successor(1337)
402 end
403
404 simgrid.platform(arg[1] or "../../msg/msg_platform.xml")
405 simgrid.application(arg[2] or "../../msg/chord/chord90.xml")
406 simgrid.run()
407