X-Git-Url: http://git.annexia.org/?p=whenjobs.git;a=blobdiff_plain;f=daemon%2Fdaemon.ml;h=cf141887f37b7fe997dc43d8e1742b2971e19cd1;hp=c3e1abd33f605dc46955bc442d5e339ff7465074;hb=46130209a2535fe06801f933e164c65084119705;hpb=f35f462b83f860958da38347d30b45986b8f4774 diff --git a/daemon/daemon.ml b/daemon/daemon.ml index c3e1abd..cf14188 100644 --- a/daemon/daemon.ml +++ b/daemon/daemon.ml @@ -78,6 +78,11 @@ let rec init j d = ~proc_get_jobs ~proc_cancel_job ~proc_start_job + ~proc_get_job + ~proc_set_variables + ~proc_get_job_names + ~proc_test_variables + ~proc_ping_daemon (Rpc_server.Unix addr) Rpc.Tcp (* not TCP, this is the same as SOCK_STREAM *) Rpc.Socket @@ -106,8 +111,10 @@ and proc_set_variable (name, value) = state := Whenstate.set_variable !state name value; (* Which jobs need to be re-evaluated? *) - let jobs = Whenstate.get_dependencies !state name in - reevaluate_whenjobs jobs; + let jobs = Whenstate.get_dependencies !state [name] in + let jobnames, state' = reevaluate_whenjobs !state jobs in + let state' = run_whenjobs state' jobnames in + state := state'; `ok with @@ -161,12 +168,97 @@ and proc_cancel_job serial = and proc_start_job jobname = try let job = Whenstate.get_job !state jobname in - run_job job; + let state' = run_job !state job in + state := state'; `ok with | Not_found -> `error "job not found" | exn -> `error (Printexc.to_string exn) +and proc_get_job serial = + try + let serial = big_int_of_string serial in + let pid = BigIntMap.find serial !serialmap in + let job, dir, serial, start_time = IntMap.find pid !runningmap in + { Whenproto_aux.job_name = job.job_name; + job_serial = string_of_big_int serial; + job_tmpdir = dir; job_start_time = Int64.of_float start_time } + with + | Not_found -> failwith "job not found" + | exn -> failwith (Printexc.to_string exn) + +and proc_set_variables vars = + try + let vars = Array.map ( + fun { Whenproto_aux.sv_name = name; sv_value = value } -> + name, variable_of_rpc value + ) vars in + let vars = Array.to_list vars in + + if !debug then + Syslog.notice "remote call: set_variables (%s)" + (String.concat " " + (List.map ( + fun (name, value) -> + sprintf "%s=%s" name (string_of_variable value) + ) vars)); + + List.iter (fun (name, _) -> check_valid_variable_name name) vars; + + (* Update all the variables atomically. *) + let s = List.fold_left ( + fun s (name, value) -> Whenstate.set_variable s name value + ) !state vars in + state := s; + + (* Which jobs need to be re-evaluated? *) + let jobs = Whenstate.get_dependencies !state (List.map fst vars) in + let jobnames, state' = reevaluate_whenjobs !state jobs in + let state' = run_whenjobs state' jobnames in + state := state'; + + `ok + with + Failure msg -> `error msg + +and proc_get_job_names () = + Array.of_list (Whenstate.get_job_names !state) + +and proc_test_variables vars = + (* This is the same as proc_set_variables, except that it doesn't + * update the state, it just returns the jobs that *would* run if + * these variables were set to these values. + *) + let vars = Array.map ( + fun { Whenproto_aux.sv_name = name; sv_value = value } -> + name, variable_of_rpc value + ) vars in + let vars = Array.to_list vars in + + if !debug then + Syslog.notice "remote call: test_variables (%s)" + (String.concat " " + (List.map ( + fun (name, value) -> + sprintf "%s=%s" name (string_of_variable value) + ) vars)); + + List.iter (fun (name, _) -> check_valid_variable_name name) vars; + + (* Update all the variables atomically. *) + let state = List.fold_left ( + fun s (name, value) -> Whenstate.set_variable s name value + ) !state vars in + + (* Which jobs WOULD be re-evaluated? *) + let jobs = Whenstate.get_dependencies state (List.map fst vars) in + let jobnames, _ = reevaluate_whenjobs state jobs in + + (* Return the names. *) + Array.of_list jobnames + +and proc_ping_daemon () = `ok + (* Reload the jobs file. *) and reload_file () = let file = sprintf "%s/jobs.cmo" !jobsdir in @@ -192,49 +284,56 @@ and reload_file () = | exn -> failwith (Printexc.to_string exn) in + let s = Whenstate.copy_prev_state !state s in state := s; (* Re-evaluate all when jobs. *) - reevaluate_whenjobs ~onload:true (Whenstate.get_whenjobs !state); + let jobs = Whenstate.get_whenjobs !state in + let jobnames, state' = reevaluate_whenjobs ~onload:true !state jobs in + let state' = run_whenjobs state' jobnames in + state := state'; (* Schedule the next every job to run. *) schedule_next_everyjob () (* Re-evaluate each when-statement job, in a loop until we reach - * a fixpoint. Run those that need to be run. + * a fixpoint. Return the list of job names that should run and + * the updated state. *) -and reevaluate_whenjobs ?onload jobs = - let rec loop set jobs = - let set' = +and reevaluate_whenjobs ?onload state jobs = + let rec loop (set, state) jobs = + let set', state' = List.fold_left ( - fun set job -> + fun (set, state) job -> let r, state' = - try Whenstate.evaluate_whenjob ?onload !state job + try Whenstate.evaluate_whenjob ?onload state job with Invalid_argument err | Failure err -> Syslog.error "error evaluating job %s (at %s): %s" job.job_name (Camlp4.PreCast.Ast.Loc.to_string job.job_loc) err; - false, !state in - - state := state'; + false, state in if !debug then Syslog.notice "evaluate %s -> %b\n" job.job_name r; - if r then StringSet.add job.job_name set else set - ) set jobs in + (if r then StringSet.add job.job_name set else set), state' + ) (set, state) jobs in + (* reached a fixpoint? *) if StringSet.compare set set' <> 0 then - loop set' jobs + loop (set', state') jobs else - set' + (set', state') in - let set = loop StringSet.empty jobs in + let set, state = loop (StringSet.empty, state) jobs in let jobnames = StringSet.elements set in (* Ensure the jobs always run in predictable (name) order. *) let jobnames = List.sort compare_jobnames jobnames in + jobnames, state +and run_whenjobs state jobnames = (* Run the jobs. *) - List.iter run_job (List.map (Whenstate.get_job !state) jobnames) + let jobs = List.map (Whenstate.get_job state) jobnames in + List.fold_left run_job state jobs (* Schedule the next every-statement job to run, if there is one. We * look at the every jobs, work out the time that each must run at, @@ -299,7 +398,8 @@ and schedule_next_everyjob () = let t_diff = if t_diff < 0. then 0. else t_diff in let run_jobs () = delete_timer_group (); (* Delete the timer. *) - List.iter run_job jobs; + let state' = List.fold_left run_job !state jobs in + state := state'; schedule_next_everyjob () in Unixqueue.weak_once esys g t_diff run_jobs; @@ -319,67 +419,101 @@ and delete_timer_group () = Unixqueue.clear esys g; timer_group := None -and run_job job = +and run_job state job = (* Increment JOBSERIAL. *) - let serial = - match Whenstate.get_variable !state "JOBSERIAL" with + let serial, state = + match Whenstate.get_variable state "JOBSERIAL" with | T_int serial -> let serial = succ_big_int serial in - state := Whenstate.set_variable !state "JOBSERIAL" (T_int serial); - serial + let state' = Whenstate.set_variable state "JOBSERIAL" (T_int serial) in + serial, state' | _ -> assert false in - Syslog.notice "running %s (JOBSERIAL=%s)" - job.job_name (string_of_big_int serial); - - (* Create a temporary directory. The current directory of the job - * will be in this directory. The directory is removed when the - * child process exits. + (* Call the pre-condition script. Note this may decide not to run + * the job by returning false. *) - let dir = tmpdir () in - - let pid = fork () in - if pid = 0 then ( (* child process running the job *) - chdir dir; - - (* Set environment variables corresponding to each variable. *) - List.iter - (fun (name, value) -> putenv name (string_of_variable value)) - (Whenstate.get_variables !state); - - (* Set the $JOBNAME environment variable. *) - putenv "JOBNAME" job.job_name; - - (* Create a temporary file containing the shell script fragment. *) - let script = dir // "script.sh" in - let chan = open_out script in - fprintf chan "set -e\n"; (* So that jobs exit on error. *) - output_string chan job.job_script.sh_script; - close_out chan; - chmod script 0o700; - - let shell = try getenv "SHELL" with Not_found -> "/bin/sh" in - - (* Set output to file. *) - let output = dir // "output.txt" in - let fd = openfile output [O_WRONLY; O_CREAT; O_TRUNC; O_NOCTTY] 0o600 in - dup2 fd stdout; - dup2 fd stderr; - close fd; - - (* Execute the shell script. *) - (try execvp shell [| shell; "-c"; script |]; - with Unix_error (err, fn, _) -> - Syslog.error "%s failed: %s: %s" fn script (error_message err) + let pre_condition () = + match job.job_pre with + | None -> true + | Some pre -> + let rs = ref [] in + IntMap.iter ( + fun pid (job, _, serial, start_time) -> + let r = { pirun_job_name = job.job_name; + pirun_serial = serial; + pirun_start_time = start_time; + pirun_pid = pid } in + rs := r :: !rs + ) !runningmap; + let preinfo = { + pi_job_name = job.job_name; + pi_serial = serial; + pi_variables = Whenstate.get_variables state; + pi_running = !rs; + } in + pre preinfo + in + if pre_condition () then ( + Syslog.notice "running %s (JOBSERIAL=%s)" + job.job_name (string_of_big_int serial); + + (* Create a temporary directory. The current directory of the job + * will be in this directory. The directory is removed when the + * child process exits. + *) + let dir = tmpdir () in + + let pid = fork () in + if pid = 0 then ( (* child process running the job *) + chdir dir; + + (* Set environment variables corresponding to each variable. *) + List.iter + (fun (name, value) -> putenv name (string_of_variable value)) + (Whenstate.get_variables state); + + (* Set the $JOBNAME environment variable. *) + putenv "JOBNAME" job.job_name; + + (* Create a temporary file containing the shell script fragment. *) + let script = dir // "script.sh" in + let chan = open_out script in + fprintf chan "set -e\n"; (* So that jobs exit on error. *) + output_string chan job.job_script.sh_script; + close_out chan; + chmod script 0o700; + + let shell = try getenv "SHELL" with Not_found -> "/bin/sh" in + + (* Set output to file. *) + let output = dir // "output.txt" in + let fd = openfile output [O_WRONLY; O_CREAT; O_TRUNC; O_NOCTTY] 0o600 in + dup2 fd stdout; + dup2 fd stderr; + close fd; + + (* Execute the shell script. *) + (try execvp shell [| shell; "-c"; script |]; + with Unix_error (err, fn, _) -> + Syslog.error "%s failed: %s: %s" fn script (error_message err) + ); + _exit 1 ); - _exit 1 - ); - (* Remember this PID, the job and the temporary directory, so we - * can clean up when the child exits. - *) - runningmap := IntMap.add pid (job, dir, serial, time ()) !runningmap; - serialmap := BigIntMap.add serial pid !serialmap + (* Remember this PID, the job and the temporary directory, so we + * can clean up when the child exits. + *) + runningmap := IntMap.add pid (job, dir, serial, time ()) !runningmap; + serialmap := BigIntMap.add serial pid !serialmap; + + state + ) + else ( + Syslog.notice "not running %s (JOBSERIAL=%s) because pre() condition returned false" + job.job_name (string_of_big_int serial); + + state + ) and tmpdir () = let chan = open_in "/dev/urandom" in @@ -400,15 +534,15 @@ and handle_sigchld _ = let job, dir, serial, time = IntMap.find pid !runningmap in runningmap := IntMap.remove pid !runningmap; serialmap := BigIntMap.remove serial !serialmap; - cleanup_job job dir serial time status + post_job job dir serial time status ) with Unix_error _ | Not_found -> () -and cleanup_job job dir serial time status = - (* If there is a cleanup function, run it. *) - (match job.job_cleanup with +and post_job job dir serial time status = + (* If there is a post function, run it. *) + (match job.job_post with | None -> () - | Some cleanup -> + | Some post -> let code = match status with | WEXITED c -> c @@ -421,12 +555,12 @@ and cleanup_job job dir serial time status = res_output = dir // "output.txt"; res_start_time = time } in - try cleanup result + try post result with | Failure msg -> - Syslog.error "job %s cleanup function failed: %s" job.job_name msg + Syslog.error "job %s post function failed: %s" job.job_name msg | exn -> - Syslog.error "job %s cleanup function exception: %s" + Syslog.error "job %s post function exception: %s" job.job_name (Printexc.to_string exn) );