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

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