(* The state. *)
let state = ref Whenstate.empty
-(* Jobs that are running; map of PID -> (job, other data). Note that
- * the job may no longer exist *OR* it may have been renamed,
+(* Jobs that are running: a map of PID -> (job, tmpdir, serial, start_time).
+ * Note that the job may no longer exist *OR* it may have been renamed,
* eg. if the jobs file was reloaded.
*)
-let running = ref IntMap.empty
+let runningmap = ref IntMap.empty
+
+(* Serial numbers of running jobs. Map of serial -> PID (in runningmap). *)
+let serialmap = ref BigIntMap.empty
(* Was debugging requested on the command line? *)
let debug = ref false
~proc_get_variable
~proc_get_variable_names
~proc_exit_daemon
+ ~proc_get_jobs
+ ~proc_cancel_job
+ ~proc_start_job
(Rpc_server.Unix addr)
Rpc.Tcp (* not TCP, this is the same as SOCK_STREAM *)
Rpc.Socket
if !debug then Syslog.notice "remote call: set_variable %s" name;
try
- (* Don't permit certain names. *)
- if name = "JOBSERIAL" then
- failwith "JOBSERIAL variable cannot be set";
-
- let len = String.length name in
- if len = 0 then
- failwith "variable name is an empty string";
- if name.[0] <> '_' && not (isalpha name.[0]) then
- failwith "variable name must start with alphabetic character or underscore";
-
- let rec loop i =
- if i >= len then ()
- else if name.[i] <> '_' && not (isalnum name.[i]) then
- failwith "variable name contains non-alphanumeric non-underscore character"
- else loop (i+1)
- in
- loop 1;
+ check_valid_variable_name name;
let value = variable_of_rpc value in
state := Whenstate.set_variable !state name value;
server := None;
`ok
+and proc_get_jobs () =
+ let running = Array.of_list (IntMap.values !runningmap) in
+ Array.map (
+ fun (job, dir, serial, start_time) ->
+ { 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 }
+ ) running
+
+and proc_cancel_job serial =
+ try
+ let serial = big_int_of_string serial in
+ let pid = BigIntMap.find serial !serialmap in
+ kill pid 15;
+ `ok
+ with
+ | Not_found -> `error "job not found"
+ | exn -> `error (Printexc.to_string exn)
+
+and proc_start_job jobname =
+ try
+ let job = Whenstate.get_job !state jobname in
+ run_job job;
+ `ok
+ with
+ | Not_found -> `error "job not found"
+ | exn -> `error (Printexc.to_string exn)
+
(* Reload the jobs file. *)
and reload_file () =
let file = sprintf "%s/jobs.cmo" !jobsdir in
Unixqueue.clear esys g;
timer_group := None
-and string_of_time_t t =
- let tm = gmtime t in
- sprintf "%04d-%02d-%02d %02d:%02d:%02d UTC"
- (1900+tm.tm_year) (1+tm.tm_mon) tm.tm_mday
- tm.tm_hour tm.tm_min tm.tm_sec
-
and run_job job =
- let () =
- (* Increment JOBSERIAL. *)
- let serial =
- 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
- | _ -> assert false in
-
- Syslog.notice "running %s (JOBSERIAL=%s)"
- job.job_name (string_of_big_int serial) in
+ (* Increment JOBSERIAL. *)
+ let serial =
+ 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
+ | _ -> 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
putenv "JOBNAME" job.job_name;
(* Create a temporary file containing the shell script fragment. *)
- let script = dir // "script" in
+ 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;
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, _) ->
(* Remember this PID, the job and the temporary directory, so we
* can clean up when the child exits.
*)
- running := IntMap.add pid (job, dir) !running
+ runningmap := IntMap.add pid (job, dir, serial, time ()) !runningmap;
+ serialmap := BigIntMap.add serial pid !serialmap
and tmpdir () =
let chan = open_in "/dev/urandom" in
let pid, status = waitpid [WNOHANG] 0 in
if pid > 0 then (
(* Look up the PID in the running jobs map. *)
- let job, dir = IntMap.find pid !running in
- running := IntMap.remove pid !running;
- cleanup_job job dir
+ let job, dir, serial, time = IntMap.find pid !runningmap in
+ runningmap := IntMap.remove pid !runningmap;
+ serialmap := BigIntMap.remove serial !serialmap;
+ post_job job dir serial time status
)
with Unix_error _ | Not_found -> ()
-and cleanup_job job dir =
+and post_job job dir serial time status =
+ (* If there is a post function, run it. *)
+ (match job.job_post with
+ | None -> ()
+ | Some post ->
+ let code =
+ match status with
+ | WEXITED c -> c
+ | WSIGNALED s | WSTOPPED s -> 1 in
+ let result = {
+ res_job_name = job.job_name;
+ res_serial = serial;
+ res_code = code;
+ res_tmpdir = dir;
+ res_output = dir // "output.txt";
+ res_start_time = time
+ } in
+ try post result
+ with
+ | Failure msg ->
+ Syslog.error "job %s post function failed: %s" job.job_name msg
+ | exn ->
+ Syslog.error "job %s post function exception: %s"
+ job.job_name (Printexc.to_string exn)
+ );
+
(* This should be safe because the path cannot contain shell metachars. *)
let cmd = sprintf "rm -rf '%s'" dir in
ignore (Sys.command cmd)
git.annexia.org / whenjobs.git / blobdiff