AI/sml_inspection/smlParallel.sml

232 fun wid_dir pd wid = pd ^ "/" ^ its wid
233 fun widin_file pd wid = wid_dir pd wid ^ "/in"
234 fun widarg_file pd wid = wid_dir pd wid ^ "/arg"
235 fun widout_file pd wid = wid_dir pd wid ^ "/out"
236 fun widscript_file pd wid = wid_dir pd wid ^ "/script" ^ its wid ^ ".sml"
237 fun param_file pd = pd ^ "/param"
238 fun result_file pd wid = wid_dir pd wid ^ "/result"
244 fun worker_listen pd wid f =
245   if exists_file (widin_file pd wid) then
246     let val s = hd (readl_rm (widin_file pd wid)) in
248         (f (); worker_listen pd wid f)
250   else (mini_sleep (); worker_listen pd wid f)
256 fun boss_stop_workers pd threadl =
260     fun send_stop wid = writel_atomic (widin_file pd wid) ["stop"]
271 fun boss_end pd threadl completedl =
274     val _ = boss_stop_workers pd threadl
290 fun send_job pd arglv warg (wid,job) =
293   warg (widarg_file pd wid) (Vector.sub (arglv,job));
294   writel_atomic (widin_file pd wid) [its job]
297 fun boss_send pd threadl rr arglv warg (pendingl,runningl,completedl) =
307     app (send_job pd arglv warg) runningl_new;
308     boss_collect pd threadl rr arglv warg
312 and boss_collect pd threadl rr arglv warg (pendingl,runningl,completedl) =
314     then boss_end pd threadl completedl
318       if not (exists_file (widout_file pd wid)) then NONE else
321       remove_file (widout_file pd wid);
331       (mini_sleep (); boss_collect pd threadl rr arglv warg prc_new)
332     else boss_send pd threadl rr arglv warg prc_new
339 fun boss_start_worker pd code_of wid =
341   writel (widscript_file pd wid) (code_of wid);
342   smlOpen.run_buildheap (wid_dir pd wid) false (widscript_file pd wid)
348 fun boss_wait_upl pd widl =
349   let fun is_up wid = hd (readl (widout_file pd wid)) = "up"
353     then app (remove_file o (widout_file pd)) widl
354     else (mini_sleep (); boss_wait_upl pd widl)
357 fun clean_parallel_dirs pd widl =
359     (mkDir_err (wid_dir pd wid);
360      app remove_file [widin_file pd wid, widout_file pd wid])
362     mkDir_err pd; app f widl
367     val pd = #parallel_dir es
368     val param = #read_param es (param_file pd)
371         val arg = (#read_arg es) (widarg_file pd wid)
374         (#write_result es) (result_file pd wid) r;
375         writel_atomic (widout_file pd wid) ["done"]
378     writel_atomic (widout_file pd wid) ["up"];
379     worker_listen pd wid f
397     val pd = #parallel_dir es
398     val _ = clean_parallel_dirs pd widl
400     val _ = #write_param es (param_file pd) param
402     fun rr wid = #read_result es (result_file pd wid)
407       boss_start_worker pd (code_of_extspec es) wid, attrib)
410     boss_wait_upl pd widl;
412     boss_send pd threadl rr arglv warg (pendingl,[],[])