Home My Page Projects Code Snippets Project Openings SML/NJ
Summary Activity Forums Tracker Lists Tasks Docs Surveys News SCM Files

SCM Repository

[smlnj] Annotation of /sml/trunk/src/cm/concur/unix-servers.sml
ViewVC logotype

Annotation of /sml/trunk/src/cm/concur/unix-servers.sml

Parent Directory Parent Directory | Revision Log Revision Log


Revision 486 - (view) (download)

1 : blume 464 (*
2 :     * Handling compile-servers under Unix- (and Unix-like) operating systems.
3 :     *
4 :     * This is still rather crude and not very robust.
5 :     *
6 :     * (C) 1999 Lucent Technologies, Bell Laboratories
7 :     *
8 :     * Author: Matthias Blume (blume@kurims.kyoto-u.ac.jp)
9 :     *)
10 :     structure Servers :> SERVERS = struct
11 :    
12 : blume 484 structure P = Posix
13 :    
14 : blume 464 type pathtrans = (string -> string) option
15 : blume 484 datatype server = S of { id: int,
16 :     name: string,
17 : blume 464 proc: Unix.proc,
18 :     pt: pathtrans,
19 : blume 484 pref: int,
20 :     decommissioned: bool ref }
21 : blume 464
22 : blume 484 fun servId (S { id, ... }) = id
23 :     fun decommission (S { decommissioned, ... }) = decommissioned := true
24 :     fun decommissioned (S { decommissioned = d, ... }) = !d
25 :     fun servName (S { name, ... }) = name
26 :     fun servPref (S { pref, ... }) = pref
27 :     fun servPT (S { pt, ... }) = pt
28 :     fun servProc (S { proc, ... }) = proc
29 :     val servIns = #1 o Unix.streamsOf o servProc
30 :     val servOuts = #2 o Unix.streamsOf o servProc
31 :    
32 :     val newId = let
33 :     val r = ref 0
34 :     in
35 :     fn () => let val i = !r in r := i + 1; i end
36 :     end
37 : blume 464 val enabled = ref false
38 :    
39 : blume 486 val idle = ref ([]: server list)
40 :     val someIdle = ref (Concur.pcond ())
41 :    
42 : blume 484 local
43 :     val nservers = ref 0
44 : blume 485 val all = ref (IntMap.empty: server IntMap.map)
45 : blume 484 in
46 :     fun noServers () = !nservers = 0
47 : blume 485 fun allServers () = IntMap.listItems (!all)
48 : blume 484 fun addServer s = let
49 :     val ns = !nservers
50 :     in
51 :     nservers := ns + 1;
52 : blume 485 all := IntMap.insert (!all, servId s, s)
53 : blume 484 end
54 :     fun delServer s = let
55 :     val ns = !nservers - 1
56 :     in
57 : blume 485 all := #1 (IntMap.remove (!all, servId s));
58 : blume 486 nservers := ns;
59 :     (* If this was the last server we need to wake up
60 :     * everyone who is currently waiting to grab a server.
61 :     * The "grab"-loop will then gracefully fail and
62 :     * not cause a deadlock. *)
63 :     if ns = 0 then
64 :     (Say.dsay ["No more servers -> back to sequential mode.\n"];
65 :     Concur.signal (!someIdle))
66 :     else ()
67 : blume 484 end
68 :     end
69 :    
70 : blume 479 (* This really shouldn't be here, but putting it into SrcPath would
71 :     * create a dependency cycle. Some better structuring will fix this. *)
72 :     fun isAbsoluteDescr d =
73 :     (case String.sub (d, 0) of #"/" => true | #"%" => true | _ => false)
74 :     handle _ => false
75 : blume 464
76 : blume 479 fun fname (n, s) =
77 :     case servPT s of
78 :     NONE => n
79 :     | SOME f => if isAbsoluteDescr n then f n else n
80 :    
81 : blume 486 (* protect some code segment from sigPIPE signals... *)
82 : blume 484 fun pprotect work = let
83 :     val pipe = UnixSignals.sigPIPE
84 :     fun disable () = Signals.setHandler (pipe, Signals.IGNORE)
85 :     fun reenable sa = ignore (Signals.setHandler (pipe, sa))
86 :     in
87 :     SafeIO.perform { openIt = disable, closeIt = reenable,
88 :     work = fn _ => work (), cleanup = fn _ => () }
89 :     end
90 :    
91 : blume 486 (* Send a message to a slave. This must be sigPIPE-protected. *)
92 : blume 464 fun send (s, msg) = let
93 : blume 479 val outs = servOuts s
94 : blume 464 in
95 : blume 479 Say.dsay ["-> ", servName s, " : ", msg];
96 : blume 484 pprotect (fn () =>
97 :     (TextIO.output (outs, msg); TextIO.flushOut outs)
98 :     handle _ => ())
99 : blume 464 end
100 :    
101 :     fun show_idle () =
102 :     Say.dsay ("Idle:" ::
103 :     foldr (fn (s, l) => " " :: servName s :: l) ["\n"] (!idle))
104 :    
105 :     (* Mark a server idle; signal all those who are currently waiting for
106 :     * that...*)
107 :     fun mark_idle s =
108 :     (idle := s :: !idle;
109 :     Concur.signal (!someIdle);
110 :     Say.dsay ["Scheduler: slave ", servName s, " has become idle.\n"];
111 :     show_idle ())
112 :    
113 :     (* Grab an idle server; wait if necessary; reinitialize condition
114 :     * if taking the only server. *)
115 :     fun grab () =
116 : blume 486 (* We need to check the following every time (at least the
117 :     * "noServers" part) because it might be that all servers
118 :     * have meanwhile gone away for some reason (crashed, etc.). *)
119 :     if not (!enabled) orelse noServers () then NONE
120 :     else case !idle of
121 : blume 464 [] => (Concur.wait (!someIdle); grab ())
122 :     | [only] =>
123 :     (Say.dsay ["Scheduler: taking last idle slave (",
124 :     servName only, ").\n"];
125 :     idle := [];
126 : blume 478 someIdle := Concur.pcond ();
127 : blume 486 SOME only)
128 : blume 464 | first :: more => let
129 :     fun best (b, [], rest) = (b, rest)
130 :     | best (b, s :: r, rest) = let
131 : blume 479 val bp = servPref b
132 :     val sp = servPref s
133 : blume 464 in
134 :     if sp > bp then best (s, r, b :: rest)
135 :     else best (b, r, s :: rest)
136 :     end
137 :     val (b, rest) = best (first, more, [])
138 :     in
139 :     Say.dsay ["Scheduler: taking idle slave (",
140 :     servName b, ").\n"];
141 :     idle := rest;
142 :     show_idle ();
143 : blume 486 SOME b
144 : blume 464 end
145 :    
146 :     fun wait_status (s, echo) = let
147 : blume 479 val name = servName s
148 :     val ins = servIns s
149 : blume 464
150 :     fun unexpected l = let
151 :     fun word (w, l) = " " :: w :: l
152 :     in
153 :     Say.say ("! Unexpected response from slave " ::
154 :     name :: ":" :: foldr word ["\n"] l)
155 :     end
156 :    
157 : blume 484 fun serverExit () = let
158 :     val what =
159 :     case pprotect (fn () => Unix.reap (servProc s)) of
160 :     (P.Process.W_EXITED | P.Process.W_EXITSTATUS 0w0) =>
161 :     "shut down"
162 :     | _ => "crashed"
163 :     in
164 :     decommission s;
165 : blume 486 Say.say ["[!Slave ", name, " has ", what, ".]\n"];
166 : blume 484 delServer s
167 :     end
168 : blume 464
169 :     val show =
170 :     if echo then (fn report => Say.say (rev report))
171 :     else (fn _ => ())
172 :    
173 :     fun loop report =
174 : blume 484 if decommissioned s then false
175 :     else
176 :     (Concur.wait (Concur.inputReady ins);
177 :     case TextIO.inputLine ins of
178 :     "" => (serverExit (); false)
179 :     | line =>
180 :     (Say.dsay ["<- ", name, ": ", line];
181 :     case String.tokens Char.isSpace line of
182 :     ["SLAVE:", "ok"] =>
183 :     (mark_idle s; show report; true)
184 :     | ["SLAVE:", "error"] =>
185 :     (mark_idle s;
186 :     (* In the case of error we don't show
187 :     * the report because it will be re-enacted
188 :     * locally. *)
189 :     false)
190 :     | "SLAVE:" :: l => (unexpected l; loop report)
191 :     | _ => loop (line :: report)))
192 : blume 464 in
193 :     loop []
194 :     end
195 :    
196 :     (* Send a "ping" to all servers and wait for the "pong" responses.
197 :     * This should work for all servers, busy or no. Busy servers will
198 :     * take longer to respond because they first need to finish what
199 :     * they are doing.
200 :     * We use wait_all after we receive an interrupt signal. The ping-pong
201 :     * protocol does not suffer from the race condition that we would have
202 :     * if we wanted to only wait for "ok"s from currently busy servers.
203 :     * (The race would happen when an interrupt occurs between receiving
204 :     * "ok" and marking the corresponding slave idle). *)
205 :     fun wait_all is_int = let
206 : blume 484 val al = allServers ()
207 : blume 479 fun ping s = let
208 :     val name = servName s
209 :     val ins = servIns s
210 : blume 484 fun loop () =
211 :     if decommissioned s then ()
212 :     else
213 :     (Concur.wait (Concur.inputReady ins);
214 :     case TextIO.inputLine ins of
215 :     "" =>
216 :     (* server has gone away -> no pong *)
217 :     Say.dsay ["<-(EOF) ", name, "\n"]
218 :     | line =>
219 :     (Say.dsay ["<- ", name, ": ", line];
220 :     case String.tokens Char.isSpace line of
221 :     ["SLAVE:", "pong"] => ()
222 :     | _ => loop ()))
223 : blume 464 in
224 :     send (s, "ping\n");
225 :     loop ()
226 :     end
227 : blume 478 val si = Concur.pcond ()
228 : blume 464 in
229 :     if List.null al then ()
230 :     else (Concur.signal si;
231 :     if is_int then
232 :     Say.say
233 :     ["Waiting for attached servers to become idle...\n"]
234 :     else ());
235 :     app ping al;
236 :     idle := al;
237 :     someIdle := si
238 :     end
239 :    
240 : blume 484 fun shutdown (s, method) = let
241 :     val i = servId s
242 :     fun unidle () =
243 :     idle := #2 (List.partition (fn s' => servId s' = i) (!idle))
244 :     fun waitForExit () =
245 :     (unidle ();
246 :     ignore (wait_status (s, false));
247 :     if not (decommissioned s) then
248 :     waitForExit ()
249 :     else ())
250 : blume 464 in
251 : blume 484 method ();
252 :     waitForExit ()
253 :     end
254 : blume 464
255 : blume 484 fun stop s =
256 :     shutdown (s, fn () => send (s, "shutdown\n"))
257 : blume 464
258 : blume 484 fun kill s =
259 :     shutdown (s, fn () => Unix.kill (servProc s, P.Signal.term))
260 : blume 464
261 :     fun start { name, cmd, pathtrans, pref } = let
262 :     val p = Unix.execute cmd
263 : blume 484 val i = newId ()
264 :     val s = S { id = i, name = name,
265 :     proc = p, pt = pathtrans, pref = pref,
266 :     decommissioned = ref false }
267 : blume 464 in
268 : blume 484 if wait_status (s, false) then (addServer s; SOME s)
269 : blume 479 else NONE
270 : blume 464 end
271 :    
272 :     fun compile p =
273 : blume 486 case grab () of
274 :     NONE => false
275 :     | SOME s => let
276 :     val f = fname (p, s)
277 :     in
278 :     Say.vsay ["[(", servName s, "): compiling ", f, "]\n"];
279 :     send (s, concat ["compile ", f, "\n"]);
280 :     wait_status (s, true)
281 :     end
282 : blume 464
283 :     fun reset is_int = (Concur.reset (); wait_all is_int)
284 :    
285 :     fun startAll st = let
286 :     val l = !idle
287 :     val _ = idle := []
288 :     val tl = map (fn s => Concur.fork (fn () => st s)) l
289 :     in
290 :     SafeIO.perform { openIt = fn () => (),
291 :     closeIt = fn () => (),
292 :     work = fn () => app Concur.wait tl,
293 :     cleanup = reset }
294 :     end
295 :    
296 :     fun cd d = let
297 :     fun st s = let
298 :     val d' = fname (d, s)
299 :     in
300 :     send (s, concat ["cd ", d', "\n"]);
301 :     ignore (wait_status (s, false))
302 :     end
303 :     in
304 :     startAll st
305 :     end
306 :    
307 :     fun cm { archos, project } = let
308 :     fun st s = let
309 :     val f = fname (project, s)
310 :     in
311 :     send (s, concat ["cm ", archos, " ", f, "\n"]);
312 :     ignore (wait_status (s, false))
313 :     end
314 :     in
315 :     startAll st
316 :     end
317 :    
318 :     fun cmb { archos, root } = let
319 :     fun st s =
320 :     (send (s, concat ["cmb ", archos, " ", root, "\n"]);
321 :     ignore (wait_status (s, false)))
322 :     in
323 :     startAll st
324 :     end
325 :    
326 :     fun dirbase db = let
327 :     fun st s =
328 :     (send (s, concat ["dirbase ", db, "\n"]);
329 :     ignore (wait_status (s, false)))
330 :     in
331 :     startAll st
332 :     end
333 :    
334 :     fun enable () = enabled := true
335 :     fun disable () = enabled := false
336 :    
337 :     fun withServers f =
338 :     SafeIO.perform { openIt = enable,
339 :     closeIt = disable,
340 :     work = f,
341 :     cleanup = reset }
342 : blume 479
343 :     val name = servName
344 : blume 464 end

root@smlnj-gforge.cs.uchicago.edu
ViewVC Help
Powered by ViewVC 1.0.0