Logo AND Algorithmique Numérique Distribuée

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